【流式数据处理】Savepoint 与升级恢复：状态演化与兼容边界

第 10 篇 讲了周期性 checkpoint：barrier 对齐、Coordinator 汇总 ack、Kafka offset 写入快照。Checkpoint 解决 运行中容错；生产上还缺一类操作——在人为边界上停作业、改代码、调并行度、升 Flink 小版本，然后从同一份状态继续跑。

Savepoint 就是 Flink 为这种场景提供的 用户拥有的、可移植的一致性快照。它与 checkpoint 共用 barrier 与状态序列化管线，但在触发方式、存储格式、生命周期和兼容承诺上 deliberately 不同。

本文回答：

• Savepoint 与 Checkpoint 各自适合什么运维动作；
• 改并行度（rescale）、改算子拓扑、改 state 类型时哪些允许、哪些会报 incompatible state；
• stop --savepoint、cancel 与 claim mode 的文件归属；
• 版本升级后从 savepoint 恢复的检查清单；
• 遇到 NonRestoredStateException、key serializer 不兼容时怎么排查。

版本锚定：Flink 1.20.x；Savepoint canonical 与 native 格式在 1.15+ 并存。Flink 2.0 将移除 LEGACY claim mode（Savepoints 文档）。下文 rescue 命令以 bin/flink CLI 为准。

环境说明：本机为 Arch Linux on WSL2（内核 6.6.87.2）、i9-12900K / 32 GB，未安装 Flink 集群。CLI 示例与配置片段来自 Flink 1.20 官方文档，不伪造 savepoint 恢复日志或耗时数据。

一、Savepoint 解决什么问题

1.1 三类典型运维动作

第 10 篇 的 checkpoint 默认 num-retained=1，面向 失败自动回滚；Savepoint 面向 计划内变更。

1.2 Savepoint 的物理结构

Flink Savepoints 文档描述：Savepoint = 稳定存储上的二进制 state 文件 + 较小的元数据文件（_metadata）。元数据含指向各 operator state 文件的相对路径；canonical 格式与具体 StateBackend 解耦，native 格式则与 RocksDB 等后端绑定（1.15+）。

目录示例：

/savepoints/savepoint-00012-abc123/
  _metadata
  ...（各 operator 的 state 文件）

Savepoint 通常可整体搬迁（复制目录到新 bucket 再 -s 恢复），但文档提醒：启用 entropy injection 或含 task-owned state（如部分 sink）时可能含绝对路径，搬迁需验证。

1.3 与第 9 篇 state 类型的关系

第 9 篇 的 ValueState / ListState / 窗口 state 都是 Savepoint 里的一行行序列化字节。State 用什么 Serializer（POJO、Avro、Kryo）直接决定后续 schema evolution 是否可行（第五节）。

二、Savepoint vs Checkpoint

2.1 生命周期与所有权

Flink 文档 Checkpoints vs. Savepoints 的核心区分：

flowchart TD
  subgraph ckpt [Checkpoint 生命周期]
    C1[interval trigger] --> C2[complete]
    C2 --> C3[保留 num-retained 份]
    C3 --> C4[failover 自动恢复]
    C3 --> C5[cancel 默认删除]
  end
  subgraph sp [Savepoint 生命周期]
    S1[用户 trigger / stop] --> S2[complete]
    S2 --> S3[用户保管路径]
    S3 --> S4[新作业 -s 恢复]
    S3 --> S5[用户 dispose]
  end

2.2 能力对照表（Flink 1.20 文档）

下表摘自官方 Checkpoints vs. Savepoints 能力矩阵，✓ 表示支持，x 不支持，! 有限制：

读表要点：

• 跨 Flink 1.x → 1.y 小版本升级：优先用 canonical savepoint；unaligned checkpoint 不能用于小版本升级路径（文档矩阵）。
• Schema evolution：仅 canonical savepoint 明确 ✓；且实际还受 Serializer 类型约束（第五节）。
• Rescale：savepoint 与 checkpoint 都支持 key-group 重分布；但特定组合有 bug（第七节 FLINK-38558）。

2.3 何时仍用 Checkpoint 做恢复

• 进程 crash、TM 丢失：自动从最近 completed checkpoint 恢复，无需人工 -s。
• 同版本、同拓扑的短周期实验：externalized checkpoint RETAIN_ON_CANCELLATION 可临时充当「廉价 savepoint」，但 不承诺跨版本与改拓扑——生产发版仍应 stop --savepoint。

2.4 Native vs Canonical Savepoint

Flink 1.15+ 触发时可指定格式：

bin/flink savepoint --type native :jobId hdfs:///savepoints/
bin/flink savepoint --type canonical :jobId hdfs:///savepoints/

大状态生产发版：若长期用 RocksDB 且不发版换 backend，native 可缩短窗口；否则 canonical 更省心。

三、Operator UID 与状态映射

3.1 Savepoint 是 Operator ID → State 的 map

Savepoints 文档：Savepoint 可视为每个有 state 的算子一条映射：

Operator ID  |  State
-------------+------------------
source-id    |  Kafka split / offset state
mapper-id    |  Keyed state

无 state 的算子（如 print）不在 savepoint 里。

3.2 必须手动 uid() 的原因

DataStream<String> stream = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
    .uid("kafka-source")
    .keyBy(...)
    .process(new MyProcessFunction())
    .uid("my-process")
    .sinkTo(sink);

• 手动 uid：拓扑小改（加无 state 算子）时，有 state 算子 ID 不变，可恢复。
• 自动 ID：随 DAG 结构变；改一行 map 可能导致 ID 漂移，savepoint 整 job 恢复失败。

文档建议：所有算子都显式 uid，因为 Window 等内置算子也有 state，表面「无 state」的节点可能仍有 state。

SQL 作业若无法设 uid，savepoint 升级难度显著上升——DataStream 生产发版应优先。

3.3 拓扑变更规则（FAQ 摘要）

Flink Savepoints FAQ（A 级）：

四、Rescale 与 KeyGroup 重分布

4.1 并行度变更如何工作

Keyed state 按 KeyGroup 分片（第 8 篇 keyBy）。Savepoint / checkpoint 里每个 subtask 存自己负责的 KeyGroupRange。

Rescale 时 Flink 读取 savepoint 中 所有 KeyGroup 数据，按 新 parallelism 重新划分 range 并分配给新 subtask。这是 在线 reshuffle state，不是简单复制文件。

# 原 parallelism=2，从 savepoint 以 4 并行启动
bin/flink run -s hdfs:///savepoints/savepoint-xxx -p 4 my-job.jar

4.2 限制与注意

• Max parallelism：创建作业时 setMaxParallelism 或 parallelism.default 上限决定了 KeyGroup 总数；savepoint 恢复时 新 parallelism ≤ max parallelism，且 不建议改 max parallelism（文档 State 部分：改 max parallelism 会导致无法从旧 savepoint 恢复）。
• Non-keyed operator state：Operator state（如 Kafka source enumerator）按 subtask 序号映射；rescaling 时 Flink 尝试 redistribute list union state；自定义 operator state 需实现 ListCheckpointed 等并支持 rescaling。
• Timer：Event time / processing time 定时器随 KeyGroup 走；RocksDB + heap timer 组合在 checkpoint rescale 有已知问题（第七节）。

4.3 rescale 与 rebalance 的区别

Savepoint 路径上的 rescale 是 计划内 扩容；运行中改 -p 必须 stop-with-savepoint 再提交新 job。

4.4 算例：KeyGroup 如何重切

设 maxParallelism=128（KeyGroup 总数固定为 128），原 parallelism=2：

• Subtask 0 持有 KeyGroup \([0, 63]\)
• Subtask 1 持有 KeyGroup \([64, 127]\)

从 savepoint 以 parallelism=4 启动时，Flink 按均匀切分规则映射（具体公式见 KeyGroupRangeAssignment 源码）：

• Subtask 0 → \([0, 31]\)
• Subtask 1 → \([32, 63]\)
• Subtask 2 → \([64, 95]\)
• Subtask 3 → \([96, 127]\)

原 subtask 0 的状态文件会被 拆分读取，属于 KeyGroup 32–63 的条目移交给新 subtask 1。这个过程在 restore 阶段完成，不是运行时 rebalance record。

若新 parallelism 不能整除 KeyGroup 分布，Flink 仍保证每个 KeyGroup 恰好归属一个 subtask，但可能出现 subtask 间 KeyGroup 数量差 1——与 第 8 篇 数据倾斜话题相关：rescaling 不消除 hot key，只改变物理并行度。

4.5 maxParallelism 规划

创建 StreamExecutionEnvironment 时：

env.setMaxParallelism(256);
env.setParallelism(8);

规则：

• Savepoint 恢复时 不能 把 maxParallelism 改小；
• 把 maxParallelism 设得过大（如 32768）会增加 KeyGroup 元数据开销，但有利于未来多次 rescale；
• 常用做法：maxParallelism = 4 × 预期最大 parallelism，且取 2 的幂次方（Flink 文档建议）。

五、State Schema Evolution

5.1 什么叫 schema evolution

State Schema Evolution 文档：允许在 savepoint 中 修改 state 数据类型（如 POJO 加字段），恢复时 Flink 用 旧 Serializer 读、新 Serializer 写 做一次性迁移。

标准流程：

1. 触发 savepoint；
2. 修改代码中 state 类型（兼容变更）；
3. flink run -s ... 恢复；首次访问该 state 时迁移。

5.2 支持的类型

POJO 规则摘要：

• 可 删除 字段：旧值丢弃；
• 可 添加 字段：新字段用 default 值；
• 不可改 已有字段类型。

5.3 与 CDC / 入湖字段变更的对照

业务表加列不等于 Flink state 自动兼容——窗口里存的 MyAggregate POJO 必须按上表演进。Avro state + Schema Registry 是长生命周期聚合作业的常见选型。

六、Cancel、Stop 与 Savepoint 操作

6.1 命令对照

Stop-with-savepoint（Flink 1.15+ 推荐发版流程）：

bin/flink stop --type canonical --savepointPath hdfs:///savepoints/ :jobId

返回 savepoint 路径后，再提交新 JAR：

bin/flink run -s hdfs:///savepoints/savepoint-xxx \
  -c com.example.MyJob my-job.jar

6.2 Detached 触发

大 state 时同步 CLI 可能超时：

bin/flink savepoint :jobId hdfs:///savepoints/ -detached

客户端立即返回 trigger id，通过 REST API 轮询完成状态（Savepoints 文档）。

6.3 --allowNonRestoredState（-n）

删除有 state 的算子或改 uid 导致无法映射时：

bin/flink run -s hdfs:///savepoints/xxx -n my-job.jar

风险：跳过 unmatched state，该算子从空 state 开始——若它是计数器/去重表，可能重复或丢业务语义。文档强调：默认拓扑序 reassigned uid 可能 错配 state，必须 显式 uid 后再用 -n 跳过 确定废弃 的算子。

6.4 Claim Mode 与文件归属（1.15+）

从 savepoint 恢复时的 claim mode 决定 snapshot 文件谁删：

bin/flink run -s hdfs:///savepoints/xxx -claimMode CLAIM my-job.jar

Native RocksDB savepoint 在 CLAIM 下可能因 SST 复用延迟目录删除（Savepoints 文档 Claim mode）。

6.5 Savepoint 存储配置

默认必须配置 savepoint 目录，否则 trigger 失败：

# flink-conf.yaml
execution.checkpointing.savepoint-dir: hdfs:///flink/savepoints

或在代码中：

env.setDefaultSavepointDir("hdfs:///flink/savepoints");

要求与 checkpoint 相同：JobManager 与 TaskManager 均可读写。权限错误表现为 trigger 成功但 subtask 上传 state 失败。

生产建议：

• savepoint 目录与 checkpoint 目录 分 prefix，避免误删；
• 对象存储上对 savepoint 设 生命周期策略 仅作用于 manual/ 前缀，不要自动清理正在用于恢复的 path；
• 大 state 发版窗口用 native savepoint 写到本地 SSD 再 distcp 到 HDFS，缩短 stop 时间（需测量网络）。

6.6 同一 checkpoint 时间线上的多作业陷阱

Savepoints 文档 Attention：若原 job 在 take savepoint 后 失败回滚到 savepoint 之前的 checkpoint，而另一 job 已从该 savepoint 启动，可能出现 事务 sink 重复提交（非确定性 + 两阶段 commit）。安全做法：stop 原 job 后再从 savepoint 启新 job；或改 transactional sink 的 uid 丢弃旧事务 state。

七、升级恢复与故障排查

7.1 版本升级检查清单

1. 阅读 Release Notes 中 State Compatibility / Savepoint 章节；
2. 发版前对生产 job trigger canonical savepoint（或 stop-with-savepoint）；
3. 在 预发 用同路径 -s 恢复，跑 smoke + 对账；
4. 确认 Flink 库版本（connector、state backend）与集群一致；
5. 若从 checkpoint 而非 savepoint 升 小版本：文档矩阵允许 aligned checkpoint，但生产仍推荐 savepoint；
6. 恢复后盯 第一次 checkpoint：NO_CLAIM 下 RocksDB 须 full checkpoint 成功后再删旧 savepoint。

7.2 常见异常与含义

7.3 FLINK-38558：RocksDB + Heap Timer + Checkpoint Rescale

Jira FLINK-38558（Flink 1.20）：state.backend.rocksdb.timer-service.factory=HEAP 时，从 checkpoint rescale 可能抛 KeyGroupRange does not contain key group N；从 savepoint 恢复 正常（savepoint 不序列化 heap timer 到 RawKeyedState，FLINK-21344）。

工程建议：rescaling 发版路径用 stop-with-savepoint；或把 timer 放到 RocksDB（ROCKSDB factory）；升级前在预发验证 rescale。

7.4 Key Serializer 不兼容：没有银弹

邮件列表与 Jira 讨论（FLINK-20375 等）：execution.savepoint.ignore-unclaimed-state 不能单独解决 key serializer 变更——跳过的只是 unclaimed operator state，key 类型变了仍无法 deserialize。

可选硬路径：

1. State Processor API：SavepointWriter.fromExistingSavepoint(...).removeOperator(uid)... 去掉问题算子 state，再写新 savepoint；
2. 新 uid + -n：旧 state 丢弃，该算子重算（需业务可接受）；
3. 不从 savepoint 恢复：重新消费 Kafka（改 group 或 timestamp）并重建 state。

Flink SQL 改 GROUP BY 键触发的 key serializer 冲突，往往只能 新作业 id + 新 state 或 API 裁剪。

7.5 诊断命令与 REST

# 列出 savepoint 元数据（路径指向 _metadata 或目录）
bin/flink savepoint -d hdfs:///savepoints/savepoint-xxx  # 确认存在再删

# 作业恢复时看 Web UI → Checkpoints → History → Savepoint restore

REST：GET /jobs/:jobid/savepoints 查看 trigger 状态（detached 模式）。

日志关键词：Cannot map old state for operator、SerializerSchemaIncompatible、Unhandled key group。

7.6 故障排查流程

flowchart TD
  E[恢复失败] --> Q1{NonRestoredState?}
  Q1 -->|是| Q2{删了算子?}
  Q2 -->|是| A1["-n 或 API 删 operator"]
  Q2 -->|否| Q3{改过 uid?}
  Q3 -->|是| A2[改回 uid 或 -n]
  Q3 -->|否| A3[查日志 unmatched operator]
  Q1 -->|否| Q4{Serializer / Migration?}
  Q4 -->|是| A4[POJO/Avro 兼容? Kryo 则新 state]
  Q4 -->|否| Q5{KeyGroup 错误?}
  Q5 -->|是| A5[FLINK-38558: 换 savepoint 或 timer 配置]
  Q5 -->|否| A6[路径/权限/版本]

7.7 发版 Runbook 模板（可复制到内部 wiki）

1. 冻结配置变更；确认当前 checkpoint 成功率 100%（最近 1 小时）。
2. flink stop --type canonical --savepointPath ... :jobId；记录返回路径与触发时间。
3. 验证 savepoint 目录大小与 _metadata 存在；抽样备份到冷存储（可选）。
4. 部署新 JAR / 新 Flink 版本；预发 -s 恢复 + -claimMode CLAIM（或 NO_CLAIM 若需保留原 snapshot）。
5. 对账：Kafka lag、业务指标、sink 侧无重复（第 15 篇 EOS 验证）。
6. 第一次 checkpoint 成功后，dispose 旧 savepoint（-d）或归档。
7. 回滚预案：保留上一版 JAR + 上一 savepoint 路径至少 24h。

八、State Processor API（边界介绍）

State Processor API 允许 读/改/写 savepoint，而不启动完整流作业——例如：从 savepoint 删除废弃算子、注入初始 state、审计 KeyGroup 分布。

• 读：canonical savepoint 最完整；checkpoint/unaligned 支持有限（对照表 !）。
• 写：仅 canonical savepoint 可创建新 snapshot。

典型裁剪流程（概念示例，类名以 Flink 1.20 文档为准）：

SavepointWriter
    .newSavepoint(env, maxParallelism, new HashMapStateBackend())
    .withOperator(OperatorIdentifier.forUid("deprecated-op"), input -> { /* drop */ })
    .write(savepointPath);

本篇不展开 API 全书；遇到 单算子 state Poison 且 -n 不够用时，再查官方 State Processor API 专章。

九、与 Checkpoint / RocksDB 的衔接

flowchart LR
  RUN[运行中] -->|failover| CKPT[最近 Checkpoint]
  RUN -->|发版| SP[Savepoint]
  SP --> NEW[新 JAR / 新 parallelism]
  CKPT --> RUN
  NEW --> RUN
  SP --> RB[RocksDB 读 SST<br/>第 12 篇]

• 日常容错：checkpoint（第 10 篇）；
• 计划变更：savepoint（本篇）；
• State 物理层：RocksDB 增量、SST 布局（第 12 篇）决定 savepoint/native 快照有多慢；
• 端到端语义：改 sink 时配合 第 15 篇 丢弃 transactional state。

十、边界

本文不展开：

• Checkpoint barrier 与 unaligned 细节（第 10 篇）；
• RocksDB compaction 与 state 调优（第 12–13 篇）；
• Kubernetes 上 savepoint 存储类选型；
• Flink CDC / SQL 无 uid 作业的完整迁移 playbook。

参考资料

1. Apache Flink Documentation 1.20, Savepoints（trigger、stop、restore、claim mode、native/canonical）。A 级。
2. Apache Flink Documentation 1.20, Checkpoints vs. Savepoints（能力矩阵）。A 级。
3. Apache Flink Documentation 1.20, State Schema Evolution（POJO/Avro/Kryo）。A 级。
4. Apache Flink Documentation 1.20, State Processor API。A 级。
5. Apache Flink Documentation 1.20, Production Ready — 为算子设置 UID。A 级。
6. ASF Jira FLINK-38558（RocksDB heap timer + checkpoint rescale）。A 级（issue）。
7. ASF Jira FLINK-21344（savepoint 与 heap timer 序列化）。A 级（issue）。
8. 本系列：第 9 篇 键控状态、第 10 篇 Checkpoint、第 12 篇 RocksDB State Backend。

返回 系列目录 · 上一篇 Checkpoint 机制 · 下一篇 RocksDB State Backend 内核路径