Checkpoint与容错：任务挂了数据不丢的秘密

📋 面试题

Flink的Checkpoint机制是怎么工作的？Checkpoint和Savepoint有什么区别？

💡 先想30秒，再往下看答案

Checkpoint解决什么问题？

实时任务7x24小时运行，一定会遇到故障（机器挂了、OOM、网络抖动）。故障恢复后，怎么保证数据不丢、不重复？

答案就是Checkpoint：定期把所有算子的State和Source的消费位点（如Kafka offset）保存到持久化存储（HDFS/S3），故障恢复时从最近的Checkpoint恢复。

Checkpoint执行流程

1. JobManager 定时触发 Checkpoint（如每60秒）
2. 向所有Source注入Barrier（屏障）
3. Barrier随数据流向下游传递
4. 每个算子收到所有上游的Barrier后：
   - 暂停处理（对齐）
   - 把State快照写到持久化存储
   - 继续处理
5. 所有算子完成快照 → Checkpoint成功

Barrier对齐（核心原理）

Source[0] —— data data [Barrier] data data ——→ 算子
Source[1] —— data data data [Barrier] data ——→ 算子

算子收到Source[0]的Barrier后：
  暂停处理Source[0]的数据，等Source[1]的Barrier
  Source[1]的Barrier到达后：
  做State快照 → 继续处理

💡 Barrier对齐会增加延迟。如果某个上游很慢，下游要一直等它的Barrier。Flink 1.11引入了非对齐Checkpoint（Unaligned Checkpoint），不等对齐直接做快照，把未处理的Buffer数据也保存进Checkpoint，用空间换延迟。

关键配置

// Checkpoint配置
env.enableCheckpointing(60000);  // 每60秒
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointTimeout(180000);  // 超时3分钟
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);  // 两次间隔至少30秒
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);  // 同时最多1个

// State Backend
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));  // true=增量Checkpoint
env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink/checkpoints");