小编点评

Flink CDC的设计架构架构概要设计中，Debezium 为变更日志提供了统一的格式结构，并支持使用 JSON 和 Apache Avro 序列化消息。 Debezium 的 changelog 数据转换为 Flink SQL 可识别的 RowData 数据，从而实现端到端的一致性保障。 Debezium 支持将 Debezium JSON 和 Avro 消息解析为 INSERT / UPDATE / DELETE 消息到 Flink SQL 系统中，方便数据处理。

正文

CDC是(Change Data Capture变更数据获取)的简称。

核心思想是，监测并捕获数据库的变动（包括数据 或 数据表的插入INSERT、更新UPDATE、删除DELETE等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。

Flink CDC的设计架构

架构的概要设计如下

为什么是Flink CDC

Debezium实现变更数据的捕获，其架构图如下

Debezium官方的架构图中，是通过kafka Streams直接实现的CDC功能。而Flink相对于Kafka Streams而言，有更多的优势：

• Flink的算子与SQL模块更为成熟和易用
• Flink作业可以通过调整算子并行度的方式，轻松扩展处理能力
• Flink支持高级的状态后端(State Backends)，允许存取海量的状态数据
• Flink提供更多的Source和Sink等生态支持
• Flink的开源协议允许云厂商进行全托管的深度定制，而kafka Streams只能自行部署和运维。

Flink Changelog Stream(Flink与Debezium的数据转换)

Debezium 为变更日志提供了统一的格式结构，并支持使用 JSON 和 Apache Avro 序列化消息。

Flink 支持将 Debezium JSON 和 Avro 消息解析为 INSERT / UPDATE / DELETE 消息到 Flink SQL 系统中。在很多情况下，利用这个特性非常的有用，例如

• 将增量数据从数据库同步到其他系统
• 日志审计
• 数据库的实时物化视图
• 关联维度数据库的变更历史

Flink 还支持将 Flink SQL 中的 INSERT / UPDATE / DELETE 消息编码为 Debezium 格式的 JSON 或 Avro 消息，输出到 Kafka 等存储中。

Debezium changelog数据转换为Flink SQL可识别的RowData数据。

Flink SQL CDC端到端数据一致性保障

Flink SQL CDC + JDBC Connector(JDBC表示为Source DB库)本质上是一个Source和Sink并行度为1的Flink Stream Application，Source和Sink之间无Operator。

一致性就是业务正确性，在“流系统中间件”这个业务领域，端到端一致性就代表Exacly Once Msg Processing(简称EOMP)，即一个消息只被处理一次，造成一次效果。即使机器或软件出现故障，既没有重复数据，也不会丢数据。

流系统端到端链路较长，涉及到上游Source层、中间计算层(Flink Operator)和下游Sink层三部分，要实现端到端的一致性，需要实现以下条件：

1.上游可以replay，否则中间计算层收到消息后未计算，却发生failure而重启，消息就会丢失。

2.记录消息处理进度，并保证存储计算结果不出现重复，二者是一个原子操作，或者存储计算结果是个幂等操作，否则若先记录处理进度，再存储计算结果时发生failure，计算结果会丢失，或者是记录完计算结果再发生failure，就会replay生成多个计算结果。

3.中间计算结果高可用，应对下游在接到计算结果后发生failure，并未成功处理该结果的场景，可以考虑将中间计算结果放在高可用的DataStore里。

4.下游去重，应对下游处理完消息后发生failure，重复接收消息的场景，这种可通过给消息设置SequcenceId实现去重，或者下游实现幂等。

Flink SQL CDC用于获取数据库变更日志的Source函数是DebeziumSourceFunction，且最终返回的类型是RowData，该函数实现了CheckpointedFunction，即通过Checkpoint机制来保证发生failure时不会丢数，实现exactly once语义，这部分在函数的注释中有明确的解释。

/**
 * The {@link DebeziumSourceFunction} is a streaming data source that pulls captured change data
 * from databases into Flink.
 * 通过Checkpoint机制来保证发生failure时不会丢数，实现exactly once语义
 * <p>The source function participates in checkpointing and guarantees that no data is lost
 * during a failure, and that the computation processes elements "exactly once".
 * 注意：这个Source Function不能同时运行多个实例
 * <p>Note: currently, the source function can't run in multiple parallel instances.
 *
 * <p>Please refer to Debezium's documentation for the available configuration properties:
 * https://debezium.io/documentation/reference/1.2/development/engine.html#engine-properties</p>
 */
@PublicEvolving
public class DebeziumSourceFunction<T> extends RichSourceFunction<T> implements
  CheckpointedFunction,
  ResultTypeQueryable<T>
  {}

为实现CheckpointedFunction，需要实现以下两个方法：

public interface CheckpointedFunction {
  //做快照，把内存中的数据保存在checkpoint状态中
  void snapshotState(FunctionSnapshotContext var1) throws Exception;

  //程序异常恢复后从checkpoint状态中恢复数据
  void initializeState(FunctionInitializationContext var1) throws Exception;
}

在Flink SQL CDC是一个相对简易的场景，没有中间算子，是通过Checkpoint持久化binglog消费位移(offset)和schema变化信息的快照，来实现Exactly Once。其实就是Checkpoint的正常功能，为实现高可用，可以将StateBackend换成HDFS等存储设备。

总结

分布式系统中端到端一致性需要各个组件参与实现，Flink SQL CDC + JDBC Connector可以通过如下方法保证端到端的一致性：

• 源端是数据库的binlog日志，全量同步做Snapshot异常后可以再次做Snapshot，增量同步时，Flink SQL CDC中会记录读取的日志位移信息，也可以replay

• Flink SQL CDC作为Source组件，是通过Flink Checkpoint机制，周期性持久化存储数据库日志文件消费位移和状态等信息（StateBackend将checkpoint持久化），记录消费位移和写入目标库是一个原子操作，保证发生failure时不丢数据，实现Exactly Once。

• JDBC Sink Connecotr是通过写入时保证Upsert语义，从而保证下游的写入幂等性，实现Exactly Once。

参考资料：

《端到端一致性，流系统Spark/Flink/Kafka/DataFlow对比总结》https://zhuanlan.zhihu.com/p/77677075

《基于Flink SQL CDC的实时数据同步方案》https://developer.aliyun.com/article/777502

《Flink SQL 1.11新功能与最佳实践》https://developer.aliyun.com/article/771773

《分布式快照算法》https://zhuanlan.zhihu.com/p/53482103

《Flink SQL CDC实践以及一致性分析》https://mp.weixin.qq.com/s?__biz=MzIwNDkwMjc1OQ==&mid=2247484847&idx=1&sn=8f011c282484a13ce38b4b425654e561&scene=58&subscene=0