Checkpoint 与 Failover 怎么落地

难度与前置

难度：进阶
前置：会提交 Flink 作业；知道 Source/Sink、状态后端、Kafka offset 是什么。
读完能做什么：给一条「Kafka → 有状态算子 → 外部 Sink」链路配好 Checkpoint 与 Exactly-Once，按 playbook 做 failover 演练并用清单验收。

适用场景

当你要跑有状态的实时作业（去重、窗口聚合、维表关联），且下游要求「故障重启后不丢数、不重复写业务表」时，用 Checkpoint + 外部系统事务/幂等把语义落到可验收。

不适用：

纯无状态、可任意重放、下游可接受重复的粗统计（可 at-least-once，不必硬上 Exactly-Once）
离线批任务（用调度重跑分区即可，不必套 Flink Checkpoint）
还没弄清 Source 是否支持可重置位点 / Sink 是否支持幂等或两阶段提交

真实场景（一条）

投放实时消耗：Kafka 明细 → Flink 按 bill_id 去重 → 写 MySQL / Iceberg ODS。某次 TaskManager OOM 后自动重启，对账发现同一小时消耗多了约 3%。

排查：Checkpoint 间隔 10 分钟，但 Sink 用「普通 JDBC 追加」且未开事务/幂等；failover 后从上次 Checkpoint 重放，MySQL 里已有部分成功写入的行被再写一次。

改法：Sink 改为幂等写入（主键 upsert）或开启两阶段提交；Checkpoint 对齐与超时按链路延迟重配；用「杀 TM → 对账」演练验收。目标：重启后同窗口金额与 Kafka 重放期望一致（幂等后 diff≈0）。

怎么做（步骤清单）

1. 先定语义目标，再选机制

目标	需要什么	常见落法
At-least-once	Checkpoint 可失败重放；下游可重复	普通 Sink + 下游幂等或离线去重
Exactly-once（端到端）	Source 可重置 + Checkpoint 对齐 + Sink 事务或幂等	Kafka Source + 两阶段提交 Sink / upsert 主键
仅算子内 Exactly-once	状态与 Checkpoint 一致即可	状态后端 + Checkpoint；Sink 另议

TIP

面试/方案里说「Exactly-Once」时，必须写清边界：算子状态语义，还是含外部 Sink 的端到端。边界写不清，对账一定扯皮。

2. Checkpoint 基础配置（可复制改）

核心参数（Flink 1.15+ 思路，键名以你们版本为准）：

yaml

# flink-conf.yaml 或作业级 Configuration
execution.checkpointing.interval: 60s
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.timeout: 10min
execution.checkpointing.min-pause: 30s
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
state.backend: rocksdb
state.checkpoints.dir: hdfs:///flink/checkpoints/${job_name}
state.savepoints.dir: hdfs:///flink/savepoints/${job_name}

作业代码侧要点：

开启 Checkpoint，间隔按「状态大小 / 上游延迟 / 下游可接受重复窗口」折中（常见 30s～3min）
RocksDB 适合大状态；小状态可用 HashMap
取消作业时保留外部化 Checkpoint，便于从最近成功点恢复
大作业加 unaligned checkpoint 前先确认版本与反压场景，别盲目开

3. Source：位点必须进 Checkpoint

Kafka Source（新 Source API）要点：

消费组 / 位点与 Checkpoint 绑定；不要作业外另起脚本「手动 commit」干扰
setStartFromGroupOffsets / 指定 timestamp 仅用于首次或明确补数；日常 failover 走 Checkpoint
Topic 扩分区后确认新分区被分配且有初始位点策略

java

KafkaSource<String> source = KafkaSource.<String>builder()
  .setBootstrapServers(brokers)
  .setTopics("ods_ad_spend_event")
  .setGroupId("flink_ad_spend_dedup")
  .setStartingOffsets(OffsetsInitializer.committedOffsets(
      OffsetResetStrategy.EARLIEST))
  .setValueOnlyDeserializer(new SimpleStringSchema())
  .build();

4. Sink：幂等或两阶段提交二选一（或组合）

Sink 类型	Exactly-once 常见做法	坑
Kafka	`FlinkKafkaProducer` / KafkaSink + 事务 / 幂等 producer	事务超时 vs Checkpoint 超时要对齐
JDBC / MySQL	主键 `UPSERT` / `ON DUPLICATE KEY`	无主键 = 无法幂等
HDFS / 对象存储	StreamingFileSink + 两阶段提交	未完成 in-progress 文件勿被下游扫到
Iceberg / Hudi	引擎侧事务提交与 Checkpoint 对齐	提交间隔过大导致可见延迟

推荐默认：业务表有自然主键 → 先做 upsert 幂等；需要「文件一次可见」再上两阶段提交。

5. Failover Playbook（值班可照做）

A. TaskManager / 单 Task 失败（常见）

Flink UI → Exceptions / Task Managers：记失败时间、异常栈（OOM / NPE / 超时）
确认 Checkpoint 历史：最近一次 Completed 时间与大小
观察自动重启次数与是否回到 Running；对比 Kafka lag 是否先升后降
若反复 Fail：先 Savepoint（若还能停），缩并行或加内存，修代码后再从 Savepoint / 外部化 Checkpoint 恢复

B. JobManager / 集群级故障

从 state.checkpoints.dir 取最新完成的 Checkpoint path
用 -s hdfs://.../chk-xxx 恢复提交（或平台「从 Checkpoint 启动」）
禁止在未确认 Checkpoint 可用时改为 latest / earliest 盲消费

C. 主动变更（扩容、改逻辑）

先 Stop with Savepoint（或兼容的 soft stop）
改并行度 / 改代码（状态兼容：uid 固定、serializer 兼容）
从 Savepoint 恢复；跑通后再删旧 Savepoint 保留策略外的过期文件

bash

# 触发 Savepoint（示例）
flink savepoint ${job_id} hdfs:///flink/savepoints/${job_name}

# 从 Savepoint / Checkpoint 恢复
flink run -s hdfs:///flink/savepoints/${job_name}/savepoint-xxxx \
  -c com.example.AdSpendJob ./ad-spend-job.jar

6. 状态与 uid 契约

每个有状态算子设稳定 uid，禁止每次发版自动生成导致状态对不上
改状态结构用 State Processor API 或兼容迁移，不要指望「同名自动升级」
维表 / Broadcast 状态也要进 Checkpoint；外部缓存（如本地 Guava）不会随 Checkpoint 恢复

java

events
  .keyBy(e -> e.billId)
  .process(new DedupFunction())
  .uid("dedup-by-bill-id")
  .name("dedup-by-bill-id");

常见坑（至少牢记这些）

Checkpoint 成功 ≠ 端到端 Exactly-Once
状态恢复正确，但 Sink 已写出的部分在重放时再写一遍。必须下游幂等或事务 Sink。
Checkpoint 超时 / 失败被「调大间隔」掩盖
间隔调到 10 分钟后看起来更稳，实则故障窗口变大、重放量暴增。应查反压、状态过大、对齐等待，而不是只放宽超时。
Cancel 不保留 Checkpoint，又从 latest 消费
丢了一段已处理但未体现在外部系统的数据，或重复一大段。生产 Cancel 用保留外部化 Checkpoint 或先 Savepoint。
算子未设 uid，发版后无法从旧 Savepoint 恢复
拓扑微变导致状态重分布失败。uid 与并行度变更策略写进发版清单。
Kafka 事务超时短于 Checkpoint 间隔
两阶段提交到 Kafka 时，事务在 Checkpoint 完成前过期，出现丢写或反复失败。对齐 transaction.timeout.ms 与 Checkpoint 超时。

如何验收（可量化）

合并前 checklist，全勾才算本篇实践落地：

[ ] 文档写清语义边界：算子内 / 端到端；Sink 用幂等还是两阶段提交
[ ] Checkpoint 开启且 EXACTLY_ONCE；外部化目录可列最新成功 chk
[ ] 有状态算子均有稳定 uid；状态后端与 chk 目录进配置库
[ ] Failover 演练：杀 1 个 TM 后作业自动恢复到 Running
[ ] 演练前后对账：同业务窗口金额 / 行数在幂等键下 diff 可解释（目标 ≈0）
[ ] Kafka 消费 lag 在恢复后单调下降，无长期卡死分区
[ ] 发版走 Savepoint → 改代码 → 恢复；禁止无 Savepoint 直接停起关键逻辑作业

验收命令 / SQL 示例：

bash

# 列出最近 Checkpoint（HDFS 示例）
hdfs dfs -ls -t hdfs:///flink/checkpoints/ad_spend_dedup | head

# 演练：记下当前 chk 后重启 TM 或 cancel+从 chk 恢复，再比 lag
kafka-consumer-groups.sh --bootstrap-server ${brokers} \
  --describe --group flink_ad_spend_dedup

sql

-- 下游幂等表：按业务窗对账（应接近 Kafka 重放期望或演练前快照）
SELECT DATE_FORMAT(event_ts, '%Y-%m-%d %H:00:00') AS hr,
       COUNT(*) AS cnt,
       SUM(cost_amount) AS cost
FROM ods_ad_spend_realtime
WHERE event_ts >= '${window_start}' AND event_ts < '${window_end}'
GROUP BY 1
ORDER BY 1;

与 Kafka 事务 / 端到端正好一次的边界，见 Kafka 事务 / EOS、端到端正好一次（Checkpoint 超时与 transaction.timeout 需交叉对齐）。

Checkpoint 与 Failover 怎么落地 ​

适用场景 ​

真实场景（一条） ​

怎么做（步骤清单） ​

1. 先定语义目标，再选机制 ​

2. Checkpoint 基础配置（可复制改） ​

3. Source：位点必须进 Checkpoint ​

4. Sink：幂等或两阶段提交二选一（或组合） ​

5. Failover Playbook（值班可照做） ​

6. 状态与 uid 契约 ​

常见坑（至少牢记这些） ​

如何验收（可量化） ​

相关面试题 ​