关注并标星 3分钟秒懂大数据

每天1次，打卡阅读

更快更全面获取大数据技术博客

Flink SQL CDC + JDBC Sink Connector 组合后如何保证一致性

在Flink SQL 1.13——CDC实践与一致性分析中，我们讲解了组件各自如何保证一致性，接下来，我们分析组合后在源库异常、Flink 作业异常、目标库异常三种异常场景下如何保证端到端一致性。

• Debezium Source 对 MySQL 进行 Snapshot 时发生异常

在 Flink Task 启动后，首先会进行 MySQL 全表扫描，也就是做 Snapshot，这里有个需要注意的地方就是，在 Snapshot 阶段，在扫描全表数据时，没有可用于恢复的位点，所以无法在全表扫描阶段去执行 Checkpoint。为了不执行 Checkpoint，MySQL 的 CDC 源表会让执行中的 Checkpoint 一直等待（通过持有 checkpoint 锁实现），甚至 Checkpoint 超时（如果表超级大，扫描耗时非常长）。这块可以从 DebeziumChangeConsumer 的代码中看到：

@Override public void handleBatch(   List<ChangeEvent<SourceRecord, SourceRecord>> changeEvents,   DebeziumEngine.RecordCommitter<ChangeEvent<SourceRecord, SourceRecord>> committer) throws InterruptedException {  try {   for (ChangeEvent<SourceRecord, SourceRecord> event : changeEvents) {    SourceRecord record = event.value();    deserialization.deserialize(record, debeziumCollector);
    if (isInDbSnapshotPhase) {     if (!lockHold) {      MemoryUtils.UNSAFE.monitorEnter(checkpointLock);      lockHold = true;            //在snapshot阶段不做checkpoint      LOG.info("Database snapshot phase can't perform checkpoint, acquired Checkpoint lock.");     }     if (!isSnapshotRecord(record)) {      MemoryUtils.UNSAFE.monitorExit(checkpointLock);      isInDbSnapshotPhase = false;      LOG.info("Received record from streaming binlog phase, released checkpoint lock.");     }    }
    // emit the actual records. this also updates offset state atomically    emitRecordsUnderCheckpointLock(debeziumCollector.records, record.sourcePartition(), record.sourceOffset());   }      ...

在做 Snapshot 阶段，可能会碰到源库 MySQL 异常或者 Flink 任务本身异常，那我们分别分析下异常后如何恢复：

1. 若遇到源库 MySQL 异常，Flink Task 发现无法连接数据库异常退出，重新启动 Flink Task（或者 retry），因为没有做 snapshot 没做 checkpoint，那么会重新再做一次 Snapshot，这些全量数据最后发送到目的 MySQL，由于下游 MySQL 实现了写幂等，因此最终保持一致性。

2. 若遇到 Flink 任务异常，重新启动（或者 retry），同上面情况一样，重新做一次 Snapshot，最终也能保持一致性。

3. 若遇到目标库 MySQL 异常，同场景一一致，Flink Task 无法往目标数据库写入异常退出，在需要重新启动或 retry 后，重新做一次 Snapshot，全量数据最后发送到目的 MySQL，由于目的下游 M 有 SQL 实现了写幂等，最终保持一致性。

• Snapshot 完成后读取 binlog 时发生异常

• 在全量数据完成同步后，开始进行增量获取，此时 Flink 会进行定时 Checkpoint，将读取 binlog 的位移信息和 schema 信息存入到 StateBackend，若此时发生异常，那我们分析下异常后如何恢复：

• 若源 MySQL 异常，Flink Task 发现无法连接数据库异常退出，重新启动 Flink Task（或者 retry），将会从最近一次 Checkpoint 的数据进行恢复，由于可以读取到 mysql binlog 位移信息，实现继续同步，不会丢失数据，最终也能保持一致性。

• 若 Flink 任务异常，重新启动或 retry 后，同场景 1 一致，继续读取 binlog，能保持一致性。

• 若目的 MySQL 异常，jdbc connector 无法往目标数据库写入，cdc connector 读取到的 binlog 位移信息也不再更新，两个操作是一个原子性操作，在 Flink Task 恢复后，从最近一次 Checkpoint 进行恢复，最终保持一致性。

  总结

• 分布式系统中端到端一致性需要各个组件参与实现，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

• 再来回顾一下端到端保持一致性的条件，发现全都能满足。

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

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

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

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

Flink SQL CDC 目前存在的缺陷

• 使用正则匹配原表后（多个源端表），到目标表无法进行一对一的映射。需要逐个匹配。

• CDC source 端定义时，需要指定所有字段，目前不支持省略字段定义。

• CDC 到 KAFKA 时无法按照主键进行自动分区分发、无法指定分区键分发数据。到 KAFKA 的数据格式指定（JSON，AVRO JSON等）。

• 目标端支持需求：DB2、ADB/GreenPlum、Oracle 暂不支持。不支持 DDL同步，不支持表的创建。

• 任务管理和监控的 REST API 不完善。

以上就是Flink SQL CDC + JDBC Sink Connector 组合后如何保证一致性讲解内容！觉得好的，点赞，在看，分享三连击，谢谢！！!

End

 找各类大数据技术文章和面经，就来

<3分钟秒懂大数据>

随时更新互联网大数据组件内容

专为学习者提供技术博文

快和身边的小伙伴一起关注我们吧！

作者简介：逆流而上Mr李，秋招7offer,CSDN博客：https://blog.csdn.net/weixin_38201936