java

5.3 Flink Checkpoint 与故障恢复

Flink Checkpoint 机制完整解析：Barrier 对齐与非对齐快照、CheckpointConfig 配置参数、Checkpoint 超时与并发控制、增量 Checkpoint 与外部化检查点、Savepoint 与作业升级、以及从 Checkpoint 恢复的完整流程。

Leeting Yan 2026-11-01 4 分钟阅读 1588 字

Checkpoint 是什么

Checkpoint 是 Flink 的一致性快照机制。定期保存所有算子的状态，故障时从最近快照恢复，保证 Flink 内部状态的 Exactly-Once。端到端的结果还要看 Source 能否重放、Sink 是否幂等或支持两阶段提交——这部分在 11.1 端到端 Exactly-Once 里展开。

正常处理：
  Source → Map → KeyBy/Window → Sink
    ↓ Checkpoint 触发
  Source 收到 Barrier，快照 offset
    → 发送到 Map
  Map 收到 Barrier，快照状态
    → 发送到 Window
  Window 收到 Barrier，快照窗口状态
    → 发送到 Sink
  Sink 收到 Barrier，确认写入，快照完成

Checkpoint 配置

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 开启 Checkpoint，每 30 秒一次
env.enableCheckpointing(30000);

CheckpointConfig config = env.getCheckpointConfig();

// 语义（CheckpointingMode 用 org.apache.flink.core.execution 包的新枚举）
config.setCheckpointingConsistencyMode(
    org.apache.flink.core.execution.CheckpointingMode.EXACTLY_ONCE);
// config.setCheckpointingConsistencyMode(
//     org.apache.flink.core.execution.CheckpointingMode.AT_LEAST_ONCE);  // 性能更好

// 超时
config.setCheckpointTimeout(60000);  // 超过 60 秒仍未完成，本次 Checkpoint 被取消并记为失败

// 并发 Checkpoint
config.setMaxConcurrentCheckpoints(1);  // 同时只允许一个在跑

// 两次 Checkpoint 最小间隔（只有 maxConcurrentCheckpoints = 1 时才会生效）
config.setMinPauseBetweenCheckpoints(10000);  // 10 秒

// 外部化 Checkpoint（作业取消后保留）；ExternalizedCheckpointCleanup 是 CheckpointConfig 的嵌套枚举
config.setExternalizedCheckpointCleanup(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// 非对齐 Checkpoint（降低反压时延迟）
config.enableUnalignedCheckpoints();
// 正数 = 先尝试对齐，超过这个时长再转非对齐
config.setAlignedCheckpointTimeout(Duration.ofSeconds(30));

Barrier 对齐 vs 非对齐

对齐 Checkpoint（默认）

Source(S1) ──→ Map(M1)
Source(S2) ──→ Map(M1)

Barrier 从 S1 到达 M1，但 S2 还没有 Barrier：
  M1 暂停处理 S2 的数据，等待 S2 的 Barrier
  等待期间的数据缓存到内存
  两个 Barrier 都到齐后，快照状态，继续处理

问题：反压时，等待 Barrier 会导致 Checkpoint 延迟甚至超时。

非对齐 Checkpoint

Barrier 从 S1 到达 M1：
  M1 不等 S2 的 Barrier，立即快照自己的状态
  S2 通道上已经读入、但还没处理完的记录，连同通道一起写进本次快照（channel state）

恢复时：先把快照里的通道数据重新灌回 M1 的输入，再接后续新数据

非对齐 Checkpoint 降低反压影响，但快照更大（多出 in-flight 数据），恢复时也要多回放一段。

config.enableUnalignedCheckpoints();
// 1.20 的默认值是 0：只要开了非对齐，每次 Checkpoint 都直接以非对齐方式开始。
// 设成正数才会"先尝试对齐，超过这个时长再转非对齐"。
config.setAlignedCheckpointTimeout(Duration.ofSeconds(30));

注意两点：setAlignmentTimeout(Duration) 是旧名字，在 1.20 里只是转调 setAlignedCheckpointTimeout，新代码直接用后者。另外 streaming.api.CheckpointingMode 这个枚举本身在 1.20 已标记废弃，改用 org.apache.flink.core.execution.CheckpointingMode 配 setCheckpointingConsistencyMode，行为等价。

状态后端

HashMapStateBackend

env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");

状态存在 JVM 堆内存
快，但受堆大小限制，大状态会直接 OOM
适合能整体装进 TaskManager 堆的小状态；具体多大要按堆配置算，不是固定 100MB

EmbeddedRocksDBStateBackend

env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");

状态存在本地 RocksDB（磁盘）
慢一些，但状态可大于内存
适合大状态（GB 级）
eventd v3 使用此后端

CheckpointStorage

// 本地文件（开发/测试）
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///tmp/checkpoints"));

// HDFS/S3（生产）
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs:///flink/checkpoints"));

Savepoint

Savepoint 是手动触发的 Checkpoint，用于：

作业升级（修改代码后恢复）
集群迁移
A/B 测试

# 触发 Savepoint
flink savepoint <job-id> hdfs:///flink/savepoints

# 从 Savepoint 恢复
flink run -s hdfs:///flink/savepoints/savepoint-xxxxx -c com.example.Job job.jar

代码中指定 Savepoint 路径：

// 注意：这个方法在 StreamExecutionEnvironment 上，不在 CheckpointConfig 上
env.setDefaultSavepointDirectory("hdfs:///flink/savepoints");

恢复流程

作业故障
    ↓
JobManager 检测到 Task 失败或 TaskManager 失联
    ↓
按 failover 策略重启受影响的任务（默认整作业重启；region 模式下只重启
  失败区域及依赖其流水线结果的下游区域，blocking 交换的下游不受影响）
    ↓
从最近成功的 Checkpoint/Savepoint 获取状态
    ↓
重新调度算子到可用 TaskManager
    ↓
每个算子从状态恢复，Source 从快照里的 offset 继续消费
    ↓
作业恢复运行

监控 Checkpoint

Flink Web UI (localhost:8081) 的 Checkpoint 页面：

指标	健康值	异常
Checkpoint Duration	远低于配置的超时时间	接近超时 → 很可能失败
Checkpointed Data Size	稳定	突然增大 → 状态泄漏
Failed Checkpoints	0	> 0 → 检查状态后端/网络
Number of Buffered Records	对齐模式下接近 0	持续很大 → 反压导致对齐卡住

这些阈值只是排查起点：Duration 与状态大小、存储带宽强相关，Buffered Records 只在对齐模式里有意义，非对齐模式要看 channel state 的大小。

eventd v3 的 Checkpoint 配置

env.enableCheckpointing(30000);   // 默认 EXACTLY_ONCE
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().enableUnalignedCheckpoints();
env.getCheckpointConfig().setAlignedCheckpointTimeout(Duration.ofSeconds(30));
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///var/flink/checkpoints");

小结

概念	说明
Checkpoint	自动一致性快照
Savepoint	手动触发，用于升级/迁移
Barrier	快照标记，贯穿数据流
对齐	等待所有输入 Barrier，状态小
非对齐	立即快照，包含 in-flight 数据
HashMapStateBackend	内存状态，快但受限
RocksDBStateBackend	磁盘状态，大但慢

第 5 章结束。下一节 6.1 Flink CEP：模式匹配入门。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页