由 dawei 摘要：本文整理自顺丰大数据研发工程师覃立辉在 5月 21 日 Flink CDC Meetup 的演讲。主要内容包括：
顺丰数据集成背景
Flink CDC 实践问题与优化
未来规划
一、顺丰数据集成背景
顺丰是快递物流服务提供商，主营业务包含了时效快递、经济快递、同城配送以及冷链运输等。
运输流程背后需要一系列系统的支持，比如订单管理系统、智慧物业系统、以及很多中转场、汽车或飞机上的很多传感器，都会产生大量数据。如果需要对这些数据进行数据分析，那么数据集成是其中很重要的一步。
顺丰的数据集成经历了几年的发展，主要分为两块，一块是离线数据集成，一块是实时数据集成。离线数据集成以 DataX 为主，本文主要介绍实时数据集成方案。
2017 年，基于 Jstorm + Canal 的方式实现了第一个版本的实时数据集成方案。但是此方案存在诸多问题，比如无法保证数据的一致性、吞吐率较低、难以维护。2019 年，随着 Flink 社区的不断发展，它补齐了很多重要特性，因此基于 Flink + Canal 的方式实现了第二个版本的实时数据集成方案。但是此方案依然不够完美，经历了内部调研与实践，2022 年初，我们全面转向 Flink CDC 。
Flink 启动之后，首先读取当前的 Binlog 信息，标记为 StartOffset ，通过 select 方式将全量数据采集上来，发往下游 Kafka。全量采集完毕之后，再从 startOffset 采集增量的日志信息，发往 Kafka。最终 Kafka 的数据由 Spark 消费后写往 Hudi。
但是此架构存在以下三个问题：
全量与增量数据存在重复：因为采集过程中不会进行锁表，如果在全量采集过程中有数据变更，并且采集到了这些数据，那么这些数据会与 Binlog 中的数据存在重复；
需要下游进行 Upsert 或 Merge 写入才能剔除重复的数据，确保数据的最终一致性；
需要两套计算引擎，再加上消息队列 Kafka 才能将数据写入到数据湖 Hudi 中，过程涉及组件多、链路长，且消耗资源大。
基于以上问题，我们整理出了数据集成的核心需求：
全量增量自动切换，并保证数据的准确性。Flink + Canal 的架构能实现全量和增量自动切换，但无法保证数据的准确性；
最大限度地减少对源数据库的影响，比如同步过程中尽量不使用锁、能流控等；
能在已存在的任务中添加新表的数据采集，这是非常核心的需求，因为在复杂的生产环境中，等所有表都准备好之后再进行数据集成会导致效率低下。此外，如果不能做到任务的合并，需要起很多次任务，采集很多次 Binlog 的数据，可能会导致 DB 机器带宽被打满；
能同时进行全量和增量日志采集，新增表不能暂停日志采集来确保数据的准确性，这种方式会给其他表日志采集带来延迟；
能确保数据在同一主键 ID 下按历史顺序发生，不会出现后发生的事件先发送到下游。
Flink CDC 很好地解决了业务痛点，并且在可扩展性、稳定性、社区活跃度方面都非常优秀。
首先，它能无缝对接 Flink 生态，复用 Flink 众多 sink 能力，使用 Flink 数据清理转换的能力；
其次，它能进行全量与增量自动切换，并且保证数据的准确性；
第三，它能支持无锁读取、断点续传、水平扩展，特别是在水平扩展方面，理论上来说，给的资源足够多时，性能瓶颈一般不会出现在 CDC 侧，而是在于数据源/目标源是否能支持读/写这么多数据。
二、Flink CDC 实践问题与优化
它基于 FLIP-27 实现，核心步骤如下：
Enumerator 先将全量数据拆分成多个 SnapshotSplit，然后按照上图中第一步将 SnapshotSplit 发送给 SourceReader 执行。执行过程中会对数据进行修正来保证数据的一致性；
SnapshotSplit 读取完成后向 Enumerator 汇报已读取完成的块信息；
重复执行 (1) (2) 两个步骤，直到将全量数据读取完毕；
全量数据读取完毕之后，Enumerator 会根据之前全量完成的 split 信息， 构造一个 BinlogSplit。发送给 SourceRead 执行，读取增量日志数据。
问题一：新增表会停止 Binlog 日志流
在已存在的任务中添加新表是非常重要的需求， Flink CDC 2.0 也支持了这一功能。但是为了确保数据的一致性，Flink CDC 2.0 在新增表的流程中，需要停止 Binlog 日志流的读取，再进行新增表的全量数据读取。等新增表的全量数据读取完毕之后，再将之前停止的 Binlog 任务重新启动。这也意味着新增表会影响其他表的日志采集进度。然而我们希望全量和增量两个任务能够同时进行，为了解决这一问题，我们对 Flink CDC 进行了拓展，支持了全量和增量日志流并行读取，步骤如下：
程序启动后，在 Enumerator 中创建 BinlogSplit ，放在分配列表的第一位，分配给 SourceReader 执行增量数据采集；
与原有的全量数据采集一样，Enumerator 将全量采集切分成多个 split 块，然后将切分好的块分配给 SourceReader 去执行全量数