Apache Flink切换sink数据丢失：Checkpoint时sink缓冲数据是否存入检查点？

你的推测核心逻辑成立，丢数根源是两类Sink的状态不兼容，具体细节如下：

1. Kafka偏移量不会提前推进，但会先于Sink刷写完成被快照 #

Flink的Checkpoint流程是：

• 检查点协调器向所有Source算子发送屏障，Source收到屏障后立即快照当前消费的最新偏移量存入Checkpoint，之后屏障继续向下游传递
• 屏障流到Sink后，Sink会快照自身的状态（包括缓冲数据、未提交文件等），再向协调器确认Checkpoint完成
• 等到所有算子都确认完成后，整个Checkpoint才被标记为有效可恢复的

也就是说，CP3里保存的Kafka偏移量，已经包含了所有流入到Sink1、还没被刷写到磁盘的缓冲数据对应的偏移量，下次从CP3恢复的时候，Source只会从这个偏移量之后开始消费，不会重复发送之前的数据。

2. 两类Sink的状态完全不兼容是丢数的直接原因 #

• BucketingSink是Flink早期的文件Sink实现，它的状态存储结构是私有的，包含待刷写的缓冲数据、当前打开的文件路径、偏移等信息
• StreamingFileSink是基于两阶段提交的新实现，状态存储的是待提交的文件事务句柄，和BucketingSink的结构完全不兼容

当你用Sink1生成的CP3启动带Sink2的作业时，Flink会自动识别到Sink算子的状态类型不匹配，直接丢弃Sink的原有状态，初始化新的Sink实例。原来存在Sink1状态里的缓冲数据就直接丢失了，这就是CP3到CP4之间数据缺口的来源。

3. 切回Sink1数据恢复的原因 #

如果你没有显式删除旧Checkpoint、且初始部署Sink1时为它设置了固定的算子uid，那么即使切换成Sink2运行，后续生成的CP4等Checkpoint里依然会保留Sink1的历史状态。切回Sink1并复用原来的uid启动作业时，Flink会加载CP4中保存的Sink1历史状态，把之前缓冲的数据写出来，同时Source从CP4记录的偏移量开始消费，所以不会出现重复数据，完全匹配你观察到的现象。

注意：所有生产环境的Flink作业都建议为每个算子显式配置唯一的uid，避免拓扑变更时出现状态匹配异常、状态丢失的问题。

• 方案1：双写过渡（最稳妥，零丢数）
  先不替换原有的BucketingSink，在作业里加一个并行度和原Sink一致的StreamingFileSink，两个Sink保持相同的分桶、序列化逻辑同时写数据，运行至少2个Checkpoint周期之后，确认两个Sink的数据完全一致，再下线旧的BucketingSink。整个过程不需要停止作业，也不需要做状态恢复，完全避免状态不兼容的问题。
• 方案2：排空后做Savepoint切换
  先暂停Kafka数据源的写入，或者等所有在途数据都处理完成、Sink的所有缓冲数据都已经刷写到磁盘之后，手动触发一次Savepoint，用这个Savepoint启动替换了Sink的新作业。因为Savepoint生成的时候，Sink已经没有缓冲数据了，所以不会有状态丢失的问题。
• 方案3：偏移量回退兼容（适合无法停流的场景）
  如果你不能停止上游写入也无法加双写，那可以先把新作业的Kafka消费偏移量手动设置为CP3对应偏移量之前1-2个Checkpoint周期的位置，启动新作业时允许小范围的重复数据，后续再通过去重逻辑或者Hive分区覆盖解决重复问题。

内容的提问来源于stack exchange，提问作者preston.m.price