62.3 状态管理与容错
一句话
流处理的正确性靠状态，可用性靠 checkpoint；状态是「到目前为止的结论」，checkpoint 是「让结论在故障后仍然成立」的全局一致快照——没有状态，流处理只是逐条转发；没有 checkpoint，状态一崩就归零。
状态的两种归属
| 类型 | 归谁管 | 典型内容 | 扩容/恢复 |
|---|---|---|---|
| Keyed State | 按 key 分区，每个 key 一份 | 计数、去重集合、窗口聚合、上次登录时间 | 随 key group 迁移；按 key 恢复 |
| Operator State | 绑定算子实例（subtask） | source 的 offset、文件读取位置、Broadcast 配置 | 随并行度重分配（list/broadcast） |
Keyed State 的四种形态：
| 形态 | 存什么 | 例子 |
|---|---|---|
ValueState | 每个 key 一个值 | 用户最近一次下单时间 |
ListState | 每个 key 一个列表 | 最近 N 次行为 |
MapState | 每个 key 一个 map | 去重 ID 集合、维度缓存 |
AggregatingState | 每个 key 一个聚合中间值 | 求和、平均值、HyperLogLog |
🔑 状态大小 = key 数 × 每个 key 的状态。key 基数（用户 ID、订单 ID、设备 ID）才是状态膨胀的根源；一个「每 key 存 10KB」的作业，1 亿 key 就是 1 TB。
State Backend：堆内存还是 RocksDB
| Heap（HashMapStateBackend） | RocksDB（EmbeddedRocksDBStateBackend） | |
|---|---|---|
| 存储位置 | JVM 堆 | 本地磁盘 + 块缓存 |
| 状态上限 | 受堆/GC 限制，通常 GB 级 | TB 级 |
| 读写延迟 | 纳秒–微秒 | 微秒–毫秒（受磁盘/缓存影响） |
| 大状态 | GC 抖动、OOM | 内存可控，但序列化/反序列化开销 |
| 增量 checkpoint | 不支持（全量） | 支持（SST 文件增量） |
| 适用 | 小状态、低延迟 | 大状态、窗口/去重/维表 |
⚠️ 把 TB 级状态放堆里，最常见的结局不是 OOM，而是 Full GC 把 checkpoint 超时，作业反复失败重启，lag 越积越多。
Checkpoint：把「所有算子的状态」切成同一个版本
窗口与水位如何驱动状态清理见 62.2 窗口与水位线；输出侧的提交见 62.4 Exactly-once 语义。
Flink 的 checkpoint 用 barrier 把一条流切成「checkpoint 前」和「checkpoint 后」：
source ──barrier──> map ──barrier──> keyBy/aggregate ──barrier──> sink
│
└─ 收到 barrier：把当前状态快照到 checkpoint 目录
对齐（aligned）checkpoint 的流程：
- source 在输入流的当前位置注入 barrier；
- 算子收到某个输入的 barrier 后，暂停该输入，把后续记录放进缓冲区；
- 等所有输入的 barrier 都到齐（对齐），把当前状态快照；
- 快照写完，向 JobManager 确认；所有算子确认后，checkpoint 完成；
- 恢复时，所有算子回到同一个 checkpoint，source 从对应 offset 重放。
非对齐（unaligned）checkpoint 则不等 barrier：直接把「缓冲区里的在途数据」也写进快照，算子继续处理，不被阻塞。代价是快照更大、恢复时重放的数据更少。
代码：对齐 vs 非对齐 vs 无 checkpoint
package main
import (
"fmt"
"sort"
)
type Ev struct {
ID string
Key string
}
var p0 = []Ev{{"p0-1", "A"}, {"p0-2", "A"}, {"p0-3", "B"}, {"p0-4", "A"},
{"p0-5", "B"}, {"p0-6", "B"}, {"p0-7", "A"}, {"p0-8", "B"}}
var p1 = []Ev{{"p1-1", "A"}, {"p1-2", "B"}, {"p1-3", "B"}, {"p1-4", "A"},
{"p1-5", "A"}, {"p1-6", "B"}, {"p1-7", "A"}, {"p1-8", "B"}}
type Op struct {
aligned bool
state map[string]int
blocked map[int]bool
buffer []Ev
snap map[string]int
snapN int
processed int
buffered int
}
func newOp(aligned bool) *Op {
return &Op{aligned: aligned, state: map[string]int{}, blocked: map[int]bool{}}
}
func (o *Op) apply(e Ev) { o.state[e.Key]++; o.processed++ }
func (o *Op) event(part int, e Ev) {
if o.aligned && o.blocked[part] {
o.buffer = append(o.buffer, e)
return
}
o.apply(e)
}
func (o *Op) barrier(part int) {
if !o.aligned {
o.snap = clone(o.state)
o.snapN = o.processed
return
}
o.blocked[part] = true
if o.blocked[0] && o.blocked[1] {
o.snap = clone(o.state)
o.snapN = o.processed
o.buffered = len(o.buffer)
for _, e := range o.buffer {
o.apply(e)
}
o.buffer = nil
o.blocked = map[int]bool{}
}
}
func clone(m map[string]int) map[string]int {
c := map[string]int{}
for k, v := range m {
c[k] = v
}
return c
}
func (o *Op) add(evs []Ev) {
for _, e := range evs {
o.apply(e)
}
}
func fmtState(m map[string]int) string {
keys := make([]string, 0, len(m))
for k := range m {
keys = append(keys, k)
}
sort.Strings(keys)
s := ""
for _, k := range keys {
s += fmt.Sprintf("%s=%d ", k, m[k])
}
return s
}
func main() {
fmt.Println("=== 对齐 checkpoint（barrier 到齐才快照）===")
oa := newOp(true)
for _, e := range p0[:4] {
oa.event(0, e)
}
oa.barrier(0)
// p0 被阻塞后，e5/e6 到达 -> 进入缓冲
oa.event(0, p0[4])
oa.event(0, p0[5])
for _, e := range p1[:4] {
oa.event(1, e)
}
oa.barrier(1)
fmt.Printf(" 快照：%s(%d 条)；缓冲 %d 条；等待 p1 %d 条\n",
fmtState(oa.snap), oa.snapN, oa.buffered, 4)
// checkpoint 完成后，处理缓冲 + 剩余
oa.add(p0[6:])
oa.add(p1[4:])
fmt.Printf(" 恢复：回到快照，重放 p0[4:]+p1[4:] = 8 条，最终 %s\n", fmtState(oa.state))
fmt.Println("\n=== 非对齐 checkpoint（不等 barrier，快照含在途）===")
ou := newOp(false)
for _, e := range p0[:4] {
ou.event(0, e)
}
ou.barrier(0)
for _, e := range p0[4:6] {
ou.event(0, e)
}
for _, e := range p1[:4] {
ou.event(1, e)
}
ou.barrier(1)
fmt.Printf(" 快照：%s(%d 条)；缓冲 0；等待 0\n", fmtState(ou.snap), ou.snapN)
ou.add(p0[6:])
ou.add(p1[4:])
fmt.Printf(" 恢复：重放 p0[6:]+p1[4:] = 6 条，最终 %s\n", fmtState(ou.state))
fmt.Println("\n=== 无 checkpoint ===")
fmt.Println(" 崩溃后状态从 0 重建：A=0 B=0（丢失 16 条）")
fmt.Println("\n=== 对比 ===")
fmt.Printf("%-10s %-15s %-10s %-8s %s\n", "mode", "snapshot-events", "buffered", "replay", "final")
fmt.Printf("%-10s %-15d %-10d %-8d %s\n", "aligned", oa.snapN, oa.buffered, 8, "A=8 B=8")
fmt.Printf("%-10s %-15d %-10d %-8d %s\n", "unaligned", ou.snapN, 0, 6, "A=8 B=8")
fmt.Printf("%-10s %-15d %-10d %-8d %s\n", "none", 0, 0, 16, "A=0 B=0")
}
本机实测（macOS，Apple M1，Go 1.27.1）：
=== 对齐 checkpoint（barrier 到齐才快照）===
快照：A=5 B=3 (8 条)；缓冲 2 条；等待 p1 4 条
恢复：回到快照，重放 p0[4:]+p1[4:] = 8 条，最终 A=8 B=8
=== 非对齐 checkpoint（不等 barrier，快照含在途）===
快照：A=5 B=5 (10 条)；缓冲 0；等待 0
恢复：重放 p0[6:]+p1[4:] = 6 条，最终 A=8 B=8
=== 无 checkpoint ===
崩溃后状态从 0 重建：A=0 B=0（丢失 16 条）
=== 对比 ===
mode snapshot-events buffered replay final
aligned 8 2 8 A=8 B=8
unaligned 10 0 6 A=8 B=8
none 0 0 16 A=0 B=0
读数：
| 模式 | 快照事件 | 缓冲/在途 | 恢复重放 | 最终 | 说明 |
|---|---|---|---|---|---|
| 对齐 | 8 | 2 | 8 | A8/B8 | p0 被暂停，e5/e6 进缓冲；等 p1 barrier 到齐才快照 |
| 非对齐 | 10 | 0 | 6 | A8/B8 | 不等 barrier，把在途数据一起写进快照；快照更大 |
| 无 checkpoint | 0 | 0 | 16 | A0/B0 | 状态丢失，重放也恢复不了历史聚合 |
🔑 对齐 checkpoint 的「快」是拿阻塞换来的：p0 在等 p1 的 4 条事件期间不能处理，缓冲会形成背压。非对齐 checkpoint 用更大的快照换掉阻塞，适合「反压敏感、状态可承受」的作业。两者都能保证状态一致，区别在吞吐与恢复成本。
Checkpoint 参数：RPO 与 RTO 的取舍
| 参数 | 作用 | 调大 | 调小 |
|---|---|---|---|
interval | 多久一次 | 恢复重放多（RPO 变大）、开销小 | 开销大、RPO 小 |
timeout | 单次超时 | 慢 checkpoint 能完成，但故障发现晚 | 容易误判失败，频繁重启 |
min-pause | 两次间隔下限 | 给作业喘息，降低背压 | 更密集，开销大 |
max-concurrent | 并发 checkpoint 数 | 快，但状态版本多、IO 竞争 | 慢，但稳定 |
unaligned | 是否对齐 | 抗背压，快照大 | 快照小，可能被反压拖慢 |
incremental | RocksDB 增量 | 快照小、快 | 恢复要合并多个增量 |
经验值（量级，不是定律）：interval 10s–1min，timeout 10min，min-pause 5s，max-concurrent 1。RPO 约等于 interval（崩溃后最多重放一个 interval 的数据），RTO 约等于「加载状态 + 重放 interval 数据」的时间。
⚠️ interval 设 1s 不会让 RPO 变成 1s，反而会让 checkpoint 一直跑不完；设 1h 会让恢复重放 1 小时数据。先量「状态大小 / 快照耗时 / 恢复耗时」，再定 interval。
恢复：状态回来了，就一定能算对吗？
source 的位点提交时机见 43.6 可靠投递与幂等消费。
恢复只保证「状态回到某个一致版本」，不保证「业务结果一定对」。三个前提：
- 重放从对应 offset 开始：checkpoint 里的 offset 与状态必须是同一个版本；
- 算子是确定性的：
now()、random()、外部接口返回值不能参与状态计算，否则重放结果不同； - 输出可重放/幂等：状态恢复后可能重放一段数据，sink 的副作用要能去重（见 62.4 Exactly-once 语义）。
🔑 checkpoint 管「状态」，savepoint 管「版本」。checkpoint 是自动的、可被清理；savepoint 是手动触发的、长期保留，用于升级、迁移、A/B 回滚。没有 savepoint，作业逻辑变更后很难回到旧状态。
State TTL：状态不是越久越好
| 状态 | TTL 依据 | 太短的后果 | 太长的后果 |
|---|---|---|---|
| 去重 ID | 消息最大重投窗口（24h–7d） | 重复消息漏判 | 状态无限增长 |
| 窗口聚合 | 窗口长度 + allowed lateness | 迟到数据无法更新 | 大量已关闭窗口占内存 |
| 维度缓存 | 维表更新频率 | 读到过期维度 | 内存膨胀 |
| 用户会话 | 业务会话超时 | 会话被切断 | 长期不活跃用户占状态 |
Flink 的 StateTtlConfig 可以配「多久过期、过期后是否清理、清理策略」。注意：TTL 是逻辑过期，不是立刻物理删除；RocksDB 的 compaction 才会真正回收。
生产落地：指名 → 可抄 → 可验 → 可退
指名：Flink（RocksDB 状态后端 + checkpoint/savepoint）；Kafka Streams 用 changelog topic + 本地 state store（没有 Flink 式 barrier checkpoint，恢复靠 changelog 重放）。
可抄（Flink 1.17+，未验证：本机未安装 Flink）：
# flink-conf.yaml（节选）
execution.checkpointing.interval: 10s
execution.checkpointing.timeout: 10min
execution.checkpointing.min-pause: 5s
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.unaligned: true
state.backend: rocksdb
state.backend.incremental: true
state.checkpoints.dir: s3://lake/flink/checkpoints
state.savepoints.dir: s3://lake/flink/savepoints
# 手动 savepoint（升级/回滚前必做）
flink savepoint <job-id> s3://lake/flink/savepoints/v1
# 从 savepoint 恢复
flink run -s s3://lake/flink/savepoints/v1 <job.jar>
Kafka Streams 侧：
state.dir=/data/kafka-streams
num.standby.replicas=1
processing.guarantee=exactly_once_v2
可验：Flink Web UI 看 lastCheckpointDuration、lastCheckpointSize、lastCheckpointRestoreTimestamp；本节程序输出对齐/非对齐的快照大小与重放条数。可退：checkpoint 失败先看「是否反压 / 状态是否过大 / 存储是否慢」，不要直接关掉 checkpoint；升级前打 savepoint，回滚时从 savepoint 恢复，而不是重跑全量。
反模式
| 反模式 | 后果 |
|---|---|
| checkpoint interval 设 1s | checkpoint 永远跑不完，作业被拖死 |
| checkpoint interval 设 1h | 故障后重放 1 小时数据，RPO 不可接受 |
| 大状态放 Heap | Full GC 导致 checkpoint 超时，反复重启 |
| 状态不设 TTL | 去重集合/窗口状态无限增长，最终 OOM |
| checkpoint 目录与计算节点同盘 | 节点故障时状态和计算一起丢 |
| 只做 checkpoint 不做 savepoint | 升级/回滚没有一致版本可回 |
用 now()/random() 参与状态计算 | 恢复后结果与崩溃前不同，无法对账 |
| 恢复后不重放 offset | 状态回来了，但中间数据丢了 |
自检
Q1 Keyed State 与 Operator State 的区别是什么？各举一个例子。
▸ 答案：Keyed State 按 key 分区，每个 key 一份，随 key group 迁移，适合计数、去重、窗口聚合；Operator State 绑定算子实例，随并行度重新分配，适合 source offset、文件读取位置、broadcast 配置。Keyed State 的恢复粒度是 key，Operator State 的恢复粒度是 subtask。
Q2 对齐 checkpoint 的过程是什么？为什么要缓冲？
▸ 答案：source 注入 barrier；算子收到某个输入的 barrier 后暂停该输入，把后续记录放进缓冲区；等所有输入的 barrier 到齐后，把当前状态快照；快照完成后继续处理缓冲区。缓冲是为了保证「快照时刻」之前的数据都进入状态、之后的数据都不进入，从而得到一个全局一致的切面。
Q3 对齐与非对齐 checkpoint 的取舍是什么？本节数字说明。
▸ 答案：对齐 checkpoint 要等所有 barrier，期间快输入被阻塞，形成背压；本节 p0 缓冲 2 条、等 p1 4 条，快照 8 条事件。非对齐 checkpoint 不等 barrier，把在途数据写进快照，不阻塞；本节快照 10 条事件、恢复重放 6 条。前者快照小但可能被反压拖慢，后者抗背压但快照更大、恢复成本更高。
Q4 checkpoint interval 怎么定？RPO 与 RTO 分别怎么算？
▸ 答案：interval 由「可接受丢多少数据」与「状态快照耗时」共同决定。RPO ≈ interval（崩溃后最多重放一个 interval 的数据）；RTO ≈ 加载最新 checkpoint 状态 + 重放 interval 数据的时间。interval 太小会让 checkpoint 一直跑、开销大；太大则 RPO/RTO 都变大。先用状态大小与快照耗时做压测，再定 10s–1min 的量级。
Q5 什么时候用 Heap，什么时候用 RocksDB？
▸ 答案：状态小（GB 级以内）、延迟敏感、key 数少时用 Heap；状态大（TB 级）、key 基数高、需要增量 checkpoint 时用 RocksDB。RocksDB 的代价是序列化/反序列化与磁盘 IO，延迟比 Heap 高；Heap 的代价是 GC 与 OOM 风险。不要把「状态大」硬扛在堆里，Full GC 会先把 checkpoint 拖超时。
Q6 checkpoint 与 savepoint 的区别是什么？
▸ 答案：checkpoint 是运行时自动触发的、用于故障恢复的一致快照，可按保留策略清理；savepoint 是手动触发的、长期保留的快照，用于升级、迁移、回滚、A/B。checkpoint 关注「快速恢复」，savepoint 关注「版本可控」。生产升级前必须打 savepoint，否则新逻辑跑错时没有旧状态可回。
Q7 为什么 now() / random() 会破坏恢复后的正确性？
▸ 答案：checkpoint 保存的是「某个时刻的状态」，恢复后会从对应 offset 重放。如果状态计算依赖 now() 或随机数，重放时这些值变了，同样的输入会产生不同的状态/输出，导致恢复前后结果不一致。正确做法是把时间/随机种子作为事件字段或确定性函数传入，保证「相同输入 + 相同状态 = 相同输出」。