汽车之家基于 Flink 的数据传输平台的设计与实践-阿里云开发者社区

数据接入与传输作为打通数据系统与业务系统的一道桥梁，是数据系统与架构中不可或缺的一个重要部分。数据传输系统稳定性和准确性，直接影响整个数据系统服务的 SLA 和质量。此外如何提升系统的易用性，保证监控服务并降低系统维护成本，优雅应对灾难等问题也十分重要。

本文介绍了汽车之家实时计算团队利用 Flink 和 Flink 实时平台构建数据传输 SDK 和传输平台并不断完善的实践经验与总结。内容包括：

背景与需求

技术选型与设计 —— Why Flink?

数据传输系统的设计架构

基于 Flink 的 Binlog 接入 SDK

平台使用

总结与展望

一、背景与需求

汽车之家（下称之家）作为一家数据智能驱动的公司，天然存在着对数据的各种复杂需求，之家的数据系统负责支撑这些业务需求的开展。数据传输系统，作为其中一环，承担了各类数据导入分发的需求，支持用户订阅数据变更。随着支撑的业务扩增与需求的增加。原来的接入系统暴露出了一定的问题和不足：

缺乏有效的任务与信息管理机制，依赖人工进行任务的管理和运维，信息的统计
接入程序资源使用浪费，缺乏弹性
针对 DDL 变更问题，不能很好的处理，必要时需要人工介入
传输系统依赖的组件比较多，比如 Zookeeper，Redis 等
代码的技术债累积，代码维护成本变高

针对上述问题，我们决定开发一套新的数据传输和分发系统，一举解决上述问题。

二、技术选型与设计 —— Why Flink?

在开展新系统的开发工作之前，我们分析的可选的方案思路大体分三种:

完全自研（类似于 otter）
复用市面上的开源组件(Maxwell/Canal/Debezium)进行二次开发和整合
基于 Flink 进行组件的开发

我们规约出以下主要设计使用目标：

架构设计上要运维管理是友好的，提供高可用以及故障恢复策略，支持异地多活
架构设计上要提供强数据准确性，至少承诺 at-least-once 语义
架构设计上要对扩缩容是友好的，可以按需分配资源
功能设计上要全面的监控覆盖和完善的报警机制，支持元数据信息管理
功能设计上要对实时计算是友好的(1)
功能设计上要能完全防御 DDL 变更带来的问题

此外，在性能指标上，接入系统的延时和吞吐至少要满足所有业务常规状态下的需求。

(1) 指与实时计算平台整合的能力

方案设计与对比

依照设计思路和目标，我们整理了方案主要功能的对比表格：

(1)Flink 自带高可用和故障恢复，实时计算平台在此基础上提供更强的高可用服务
(2)良好的编码 + flink 机制即可实现 Exactly-Once
(3)实时计算平台自带任务部署管理能力
(4)实时计算平台自带完备的监控和管理

经过讨论，大家一致决定基于Flink进行新的传输平台的开发：

Flink DataStream 的编程模型和 API 在应对数据传输场景上，非常的自然与直接
Flink 在框架层面提供了一致性保证和 HA/稳定性/流量控制措施，让我们可以不必去处理这些开发上比较困难和复杂的问题，背靠框架即可较为轻松地完成相关工作
Flink 天然具备横向纵向扩容的能力，按需使用计算资源即可
完全复用了之家 Flink 实时计算平台已有的组件和能力——完备的监控报警/任务生命周期管理/异地多活/自助运维等功能

我们的 MVP 版本开发完成大约只花费了不到 3 周的时间，POC 的结果完全符合预期的性能要求和功能要求。

三、数据传输系统的设计架构

从逻辑层面来看，之家的实时数据传输平台分为 3 部分：

数据传输程序
接入任务信息管理模块
任务执行 Runtime 模块

在实现上：

数据传输程序是由固定的 Flink Jar 和 Flink SQL Codegen Service 生成的SQL Task 组成
管理模块作为一个微服务，负责与 Flink 平台组件通信，完成必要的任务管理和信息管理
执行层直接依赖 Flink 平台和 Flink 平台的集群

组件架构与交互逻辑

传输系统涉及到的组件和交互如图所示：

AutoDTS 即为传输系统的任务信息管理模块，AutoStream Core 为 Flink 实时平台核心系统，Jar Service 是 Flink 相关 SDK Jar 储存管理服务，Metastore 为 Flink 平台的元数据管理系统，Flink Client 是我们自己封装的 Submit Client，支持以 Restful 方式向 YARN/K8S 上提交作业。

AutoDTS 前端直接与用户进行交互，完成用户对任务信息的修改和任务生命周期的操作。AutoDTS 将任务信息处理后与 Flink 平台交互，每一个数据传输任务对应Flink平台唯一一个任务，同时，部分任务信息被 AutoDTS 处理，会直接在 Metastore 上完成对应流表的创建。用户直接申请并使用该 Flink 流表，进行 SQL 任务的开发。

针对不同的传输任务，AutoDTS 会委托 Core System 组织任务参数和 SQL 逻辑，并从 Jar Service 加载不同的 SDK Jar 提交到 Client 去执行，对于基于 SQL Codegen 的传输任务，Flink SQL Codegen Service 会将任务参数组织整合翻译成可执行的 Flink SQL 任务，通过 SQL 任务，我们可以直接复用平台 SQL SDKs，执行 SQL 作业。

正如前文提到的，我们最大限度复用已有组件和服务，大大降低了开发的周期。

传输任务类型与构成

之家的数据传输任务分为两种类型，接入任务与分发任务。

接入任务，负责从数据源实时接入 Changelog Stream 并处理成统一的格式写入 Kafka 中，每个表只会对用唯一个接入程序，作为公共数据资产，被下游程序进行使用和消费
分发任务，负责读取公共的 Kafka 数据，并将数据写入指定的存储中，用户根据自己的需求去使用，拥有分发任务的所有权

如图所示，接入的数据源主要有 3 种，除了 Mysql 和 SqlServer，我们还支持了 TiDB 的 Changelog(TiCDC)接入 Java Client 相关逻辑,并将我们的代码贡献到了 TiDB 社区 [1]；对于分发端，通过解析用户的任务配置，从而进行 SQL codegen 生成 Flink SQL 代码执行。

四、基于 Flink 的 Binlog 接入 SDK

在这些接入和分发 SDK 中，Binlog 接入 SDK 是比较有难度的一个，下面我们以 Binlog 接入 SDK 为例，剖析接入 SDK 的主体设计思路和开发过程。

Stage 拆解

依照 Flink 经典的 Source->Transformation->Sink，Binlog 接入任务也拆分为这三个Stage：

Binlog Source

Binlog Source 的朴素开发思路：创建一个 BinaryLogClient 并持续 fetchBinlogEvent 并进行简单的转换处理后发送到下游。在既定的设计目标中，以下问题需要认真思考：

保证 Source 端处理性能
保证 source 是可回溯的
保证 Mysql Transaction 的完整性

对于问题1，考虑到 Binlog Stream 的特殊性，我们要求 Source 的并行度为且仅能为1。且在绝大部分情况下，从 BinaryLogClient fetch BinlogEvent 不会是性能瓶颈。我们只要保证 BinaryLogClient 与 BinlogSourceFunction 的生命周期一致，二者通过有界的阻塞队列链接，分别充当生产者和消费者，同时 BinlogSourceFunction 对 BinlogEvent 尽可能少的进行逻处理，让 BinlogSourceFunction 的负担尽量减轻，从而提升 Source 阶段的性能即可。

而对于问题 2、3，则需要从 Binlog 的特性和格式来分析。众所周知，BinlogEvent 携带了唯一的 BinlogPosition。BinlogPosition 是全序的，我们可以在 trigger Checkpoint 的时候，对当前的 BinlogPosition 进行记录。但是仅仅是记录这个是不够的，如果记录了数据位置，那么下次从 Checkpoint 恢复的时候，是从当条记录开始还是当条记录的下一条记录开始呢？另一方面，我们希望发送的按照一个完整的 transaction 去发送数据给下游而非从事务中间截断发送。这里，我们就要用到 BinlogEvent 的一种特定事件——TransactionEnd 事件。

我们这里先来解决问题 2，我们要求 BinlogSourceFunction 只使用 TransactionEnd 事件的 BinlogPosition 来更新位点保存到状态中，由于 TransactionEnd 事件不是 DML 事件，不会导致下游生成数据，所以就不需要考虑之前提到的问题。

而问题3的解决需要和 Flink的Checkpoint 机制进行联动。我们当时使用的 Flink 版本是 1.9.x。在 Source 端，需要通过 CheckpointLock 来让 Source 和 Checkpoint trigger 进行配合。虽然在理解和使用上有一定的壁垒，但是 CheckppointLock 机制恰恰帮助我们达成了问题 3 的目标。我们保证了 Source 只有拿到 lock 才发送数据给下游，只有在完成一次 transaction 的数据发送后才 unlock，这样就保证了 2 个 checkpoint 之间必定是完整的 ?（ ? ∈ N ）次 transaction 的数据。另一方面，我们减小了 checkpoint trigger 的间隔（200ms~500ms），减少了 checkpoint 间的数据 transaction 的数量，加快数据 commit 的速度。

UnifiedFormatTransform

就如名字描述的，UnifiedFormatTransform 的作用是将数据转换为统一制定的数据格式。

相较于 Binlog Source 阶段，UnifiedFormatTransform 阶段不用太过担心性能问题，良好的编码和水平垂直扩容能力可以应付绝大部分性能需求。但是有一个重要的问题亟待解决，就是前面提到的功能设计目标：完全防御 DDL 带来的问题。

DDL 问题在数据同步/传输中一直是一个比较棘手的问题，带来的麻烦包括不限于，数据解析失败/错误，程序失败/重启，且恢复的成本往往很高。而其实解决这个问题的核心思路也很简单，就是在程序中就地解析 DDL 并处理 Schema 变化。为了实现这个功能，我们需要完成以下几个步骤：

内嵌 Parser，用于解析 DDL SQL
解析出现的所有 DDL，根据解析的 DDL 内容更新内置的 Schema，并更新到 Flink 状态中
生成 DDL 对应的数据发送到下游

我们这实现上参考了 Maxwell [2] 的做法，内嵌了 Antlr4 的 Mysql 文法的 g4 文件，然后自定义 listener 来完成对 Schema 的更新和 DDL 数据的生成，然后 Schema 会在 Checkpoint 触发时被保存到状态中。

完成了就地解决 DDL 的功能后，不论是简单的 Alter Table，还是复杂的 Online DDL，接入程序都可以顺利解决，利用状态从断点恢复，也不会出现 Schema 异常的问题。

Kafka Sink

Kafka Sink 阶段主要是将转换好的数据写入 Kafka中。Flink 原生为 Kafka Sink 赋予了 Exactly-Once 的能力，而我们也将这个功能利用起来，和 Source 一起，提供了开箱即用的端到端 Exactly-Once 解决方案。我们保证了 Source 按照完整的 Mysql Transaction 发送数据，同时 Sink 按照完整的 Mysql Transaction 将数据写入Kafka，对于 Transaction 敏感的场景，我们可以开启 Transactional 消费模式，来完成强 transaction 语义（而非最终一致性）的数据处理。

其他优化

此外我们还做了一些优化功能：

gtid 支持与一键主从切换
程序运行信息定期备份到外部存储
Binlog 同步任务相关的监控指标覆盖

五、平台使用

用户在传输平台，只需要完成必要配置的设定，即可完成传输任务的创建和数据的使用，比较简单。

接入任务

对于接入任务，正如我们前文提到的, 接入任务产生的数据会被作为公共资产。所以用户只需要查询需求的表的数据是否已经接入，如果已经接入，则可以直接申请使用，否则发起一次表接入申请，审批通过后会由系统自动进行操作。

分发任务

对于分发作业，需要用户进行创建，以 Iceberg 分发任务为例：

■ 字段筛选

选择出分发作业使用的已经接入到平台的数据源表字段

在选择一些任务的运行配置（如资源，运行环境）后，就可以创建并运行一个分发任务，我们可以看到对应唯一一个 Flink 平台任务 ID：

此外，我们还提供了丰富的监控查询，元数据信息查询等功能，充分利用了实时计算平台的已有组件，实现了传输系统与实时计算系统的紧密结合。

六、总结与展望

实践证明，我们选择基于 Flink 进行输出传输系统的开发，是个明智且正确的决定。在最小的开发成本下，从功能和效率及可维护性上，完全解决了之前遗留的问题，全面提升了之家接入/分发/数据订阅的效率和用户体验，也提升了我们在数据传输方面的技术能力。

最近我们在数据湖方向投入了较多的精力，传输系统目前也已经初步支持数据接入数据湖，未来希望可以不断完善相关功能，大幅提升数据湖数据接入的能力，支持用户一键入湖，加强整个数据体系的整合。

另一方面，我们看到 Flink 新版本提供了许多新功能新工具。例如 FLIP-27 Source 和 OperatorCoordinator，我们希望可以借由这两个全新的机制和工具，继续优化我们的代码，拓展相关功能。对于新推出 Upsert-Kafka，我们已经开始尝试在 Flink 计算平台上进行初步的开发和整合，希望之后将 Upsert-Kafka 与传输系统打通，继续扩展与丰富实时计算和传输的业务场景！