Flink CDC里使用增量快照，怎么读取最近一次的checkpoint?

问题分析

在使用Flink CDC时，启用了增量快照（scan.incremental.snapshot.enabled='true'）和启动模式为initial（scan.startup.mode='initial'），并且checkpoint已经写入S3。当JobManager崩溃重启后，如何读取最近一次的checkpoint？同时，用户提到出现了sink重复数据的问题。

以下是针对该问题的详细解答：

1. 如何读取最近一次的checkpoint

Flink支持从最近一次的checkpoint恢复作业，具体操作步骤如下：

1.1 确保checkpoint配置正确

• 在Flink作业中，checkpoint需要正确配置以确保能够持久化到S3。以下是一些关键参数：

  execution.checkpointing.interval: 60000 # 检查点间隔时间（毫秒）
  state.backend: rocksdb # 使用RocksDB作为状态后端
  state.checkpoints.dir: s3://your-bucket/checkpoints # checkpoint存储路径
  execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION # 保留checkpoint
  

  确保这些参数已正确设置，并且checkpoint已成功写入S3。

1.2 从checkpoint恢复作业

• 当JobManager崩溃后，可以通过以下方式从最近一次的checkpoint恢复作业：
  • 命令行方式： 使用Flink CLI提交作业时，指定-s参数指向最近一次的checkpoint路径。例如：

  ./bin/flink run -s s3://your-bucket/checkpoints/<job-id>/chk-<checkpoint-id> -d your-job.jar
  

  • Web UI方式： 在Flink Web UI中，选择“Submit Job”页面，在“Savepoint Path”字段中填入最近一次的checkpoint路径。

1.3 验证恢复是否成功

• 恢复后，Flink会从checkpoint中加载状态并继续处理数据流。您可以通过Flink Web UI或日志确认作业是否成功恢复。

2. sink重复数据问题的原因及解决方案

2.1 问题原因

• At-Least-Once语义：Flink默认提供At-Least-Once语义，这意味着在故障恢复时可能会出现重复数据。如果sink未实现幂等性或未启用Exactly-Once语义，则可能导致重复写入。
• checkpoint恢复机制：当从checkpoint恢复时，Flink会重新处理checkpoint之后的数据，这可能导致部分数据被重复处理。

2.2 解决方案

根据问题描述，以下是几种解决sink重复数据问题的方法：

方法一：启用Exactly-Once语义

• 如果您的sink支持Exactly-Once语义，可以通过以下方式启用：
  • 设置Flink的checkpoint模式为EXACTLY_ONCE：

  execution.checkpointing.mode: EXACTLY_ONCE
  

  • 确保sink实现了两阶段提交协议（Two-Phase Commit Protocol）。例如，Kafka sink支持Exactly-Once语义。

方法二：实现幂等性

• 如果sink不支持Exactly-Once语义，可以通过实现幂等性来避免重复数据。例如：
  • 在MySQL结果表中，为主键列添加唯一约束，确保重复插入不会导致数据重复。
  • 在Hologres或PostgreSQL中，使用INSERT ... ON CONFLICT语句实现幂等写入。

方法三：调整checkpoint间隔

• 减少checkpoint间隔时间可以降低重复数据的范围。例如，将execution.checkpointing.interval从60秒调整为30秒。

方法四：检查CDC源表配置

• 确保CDC源表的scan.startup.mode配置正确。如果设置为initial，Flink会在首次启动时扫描全量数据，然后切换到增量数据。如果作业频繁重启，可能会导致全量数据被重复消费。
  • 建议在首次启动后，将scan.startup.mode改为latest-offset，以避免重复消费全量数据。

3. 重要提醒

• 主键配置：在使用增量快照功能时，必须为CDC源表配置主键（Primary Key）。如果未配置主键，可能会导致报错或数据重复问题。
• MongoDB前像后像功能：如果您使用的是MongoDB CDC连接器，请确保已开启前像后像记录功能（Pre- and Post-images），否则可能导致数据不一致。
• 并发读取限制：如果启用了增量快照的并发读取功能（scan.incremental.snapshot.enabled=true），请确保server-id范围足够大以支持并发数。

总结

通过上述方法，您可以从最近一次的checkpoint恢复作业，并解决sink重复数据的问题。建议优先启用Exactly-Once语义或实现幂等性，以确保数据一致性。