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

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

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

1. 背景与需求
2. 技术选型与设计 —— Why Flink?
3. 数据传输系统的设计架构
4. 基于 Flink 的 Binlog 接入 SDK
5. 平台使用
6. 总结与展望

一、背景与需求

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

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

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

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

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

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

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

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

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

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

方案设计与对比

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

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

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

1. Flink DataStream 的编程模型和 API 在应对数据传输场景上，非常的自然与直接
2. Flink 在框架层面提供了一致性保证和 HA/稳定性/流量控制措施，让我们可以不必去处理这些开发上比较困难和复杂的问题，背靠框架即可较为轻松地完成相关工作
3. Flink 天然具备横向纵向扩容的能力，按需使用计算资源即可
4. 完全复用了之家 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 并进行简单的转换处理后发送到下游。在既定的设计目标中，以下问题需要认真思考：

1. 保证 Source 端处理性能
2. 保证 source 是可回溯的
3. 保证 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 之间必定是完整的