★ FEATURED ARTICLE 1. 把快照这件事拆开Checkpoint 里到底存了哪几样东西先说一个我见过很多次的现场作业跑了三个月某天机器故障重启结果消费位点直接退回到四十分钟前下游报表里凭空多出一批重复数据。查了半天发现 Checkpoint 一直是失败的失败日志被淹没在业务日志里没人看。这件事让我意识到一个很朴素的事实——Flink 实时计算里的 Checkpoint 不是一个配了就完事的开关它是一条完整的链路从 JobManager 的调度器到 TaskManager 的本地状态再到远端存储任何一环掉链子容错能力就等于零。Checkpoint 的本质是一次分布式的、异步的、全局一致的状态快照。注意这里面有三个关键词分布式意味着快照由成百上千个算子实例协同产生异步意味着它不能停下来等数据流必须继续跑全局一致意味着所有算子拍下的必须是同一时刻的逻辑视图不能张三拍了 10 点的数据、李四拍了 10 点半的数据。这三点凑在一起难度就上来了——既要快又要不阻塞还要保证一致性。Flink 用的是 Chandy-Lamport 算法的一个变种业界一般叫它异步屏障快照Asynchronous Barrier Snapshotting简称 ABS。理解 ABS 是理解后面所有参数、所有坑的前提。那这个快照里到底装了什么很多人以为就是算子里的那个 Map实际上远比这复杂我习惯把它分成三层来看。1.1 第一层算子的内部状态这才是大头内部状态就是算子在运行过程中自己维护的数据。写得最多的keyBy之后的ValueState、ListState、MapState还有窗口里缓存的元素、ProcessFunction里定时器的队列、join场景下两侧的缓冲全都属于这一层。这部分的体量往往决定了 Checkpoint 的大小也决定了恢复要花多久。这里有个容易被忽略的点状态是按 key group 组织的不是按 key 组织的。Flink 会把整个 key 空间切成固定数量的 key group默认max parallelism是 128也就是 128 个 key group。每个并行实例负责其中的一段。这个设计是为了让并行度可以调整——只要 key group 总数不变你从 4 并发扩到 8 并发状态只是重新分配一下不需要重算。但反过来说一旦你显式设了setMaxParallelism这个数字就成了永久契约后面改不了改了 Savepoint 就恢复不了。我在项目里一般会在第一版上线前就把最大并行度定好宁可留大一点。状态后端的选择直接改变这一层的存储形态。堆内存后端HashMapStateBackend把状态放在 JVM 堆里快照时序列化后往外传优点是读写快缺点是状态一大就容易 GC 抖动几个 GB 的状态基本就别考虑了。RocksDB 后端把状态放在本地磁盘的 LSM 树上堆里只留读缓存和写缓冲好处是能撑住几十上百 GB而且支持增量快照——因为 LSM 树的 SST 文件一旦落盘就不再修改增量 Checkpoint 只需要把新产生的 SST 文件传上去旧的文件用引用指过去就行。这个机制在状态大、变更少的场景下能把每次 Checkpoint 的数据量从几十 GB 压到几百 MB是实打实的性能差异。1.2 第二层数据源的位点决定重放从哪开始Kafka Source 的 offset、文件源的读取位置、数据库 CDC 的 binlog 位置都属于这一层。这部分数据量极小几十个字节但它决定了故障恢复后的语义边界。举个例子你的作业从 Kafka 消费offset 存在外部比如自己写 Redis和存在 Flink 状态里行为完全不同。存在外部时Checkpoint 和位点提交是两套独立机制作业挂掉重启位点可能已经被提交到很远的地方中间那段数据就永久丢了。存在 Flink 状态里现在 Source 默认就是这么干的位点和算子状态被打包进同一个快照恢复时两者严格对齐这才谈得上精确一次。所以我在做任何实时链路时第一件事就是确认 Source 的位点究竟由谁管理。这个决定比后面调什么参数都重要因为它是语义的地基。1.3 第三层算子链的元信息与在途数据元信息包括作业的拓扑结构、算子的 UID、并行度配置、序列化器版本等等。这部分是恢复时能不能对得上号的关键也是最容易在版本升级时出问题的地方——UID 变了、算子顺序变了、序列化器不兼容了快照读出来就是一锅粥。在途数据in-flight data平时不在快照里但如果你开了非对齐 Checkpoint那些卡在网络上、卡在输入缓冲区里还没来得及处理的数据也会被一并快照下来。这是后面第 2 节要细说的内容先记住一点开了非对齐快照会变大恢复时重放的数据会变多但 Checkpoint 本身会更快更稳这是个明确的天平。2. Barrier 在流里插队的过程决定了你能承受多大的乱序与背压理解了快照里有什么接下来要回答的是怎么让所有算子在同一时刻拍照。Flink 的做法很巧妙它不喊全体暂停而是往数据流里插入一个特殊标记——Barrier。这个 Barrier 就像跑道上的发令员跟着数据一起往下游流。算子收到 Barrier就知道从这往前的数据归这一轮快照。听着简单难点在于一个算子往往有多个输入通道。比如一个keyBy之后的算子上游可能接了 8 个并行实例那它就有 8 个输入。这 8 个通道的数据流速不可能完全一样Barrier 到达的时间自然有先后。先到的那个通道Barrier 后面还跟着一堆数据这些数据属于下一轮快照的范围不能现在就处理。怎么办只能先把它们缓冲起来等所有通道的 Barrier 都到齐。这个等待动作就叫对齐alignment。2.1 对齐窗口期的背压放大效应对齐听起来只是等一会儿实际影响比想象中大得多。在等待期间先到 Barrier 的通道被堵住了——数据不能处理只能往缓冲区里堆。缓冲区满了反压就沿着这条链路往上游传上游的 Source 被压住整个作业的吞吐掉下来。等最后一个通道的 Barrier 到了积压的数据才开始被处理这时候又是一波流量洪峰。问题在于这种放大效应和数据倾斜是乘数关系。假设某个 key 的数据量是平均值的 20 倍那么这个通道的处理速度就慢 20 倍Barrier 传得也慢 20 倍其他 7 个通道就得干等 20 倍的时间。状态越大、对齐时间越长、积压越多下一轮 Checkpoint 的对齐时间更长——形成恶性循环。我见过一个作业最初 Checkpoint 只要 3 秒跑了两周变成 90 秒最后直接超时失败根因就是某个维度上出现了热点 key再加上对齐的放大效应。所以当你看到 Checkpoint 时长持续上升、而状态大小没什么变化时第一反应应该是去看反压和倾斜而不是去调 Checkpoint 参数。调参数只是把症状掩盖住根因还在那里。对抗这个问题的手段有两个方向。一是治理数据倾斜加盐打散、两阶段聚合、把热点 key 单独拆出来处理这是治本。二是把 Checkpoint 模式换成AT_LEAST_ONCE或者开启非对齐 Checkpoint直接取消对齐这个动作。前者会带来重复消费后者不会丢数据但会让状态变大。2.2 关掉对齐之后非对齐 Checkpoint 的取舍非对齐 Checkpoint 的思路很直接既然等 Barrier 对齐这么痛苦那就不等了。Barrier 一到就往下游转发不用管其他通道有没有到。那些还没处理完的在途数据怎么办直接快照下来写到临时文件里恢复的时候先恢复状态再把这些在途数据重放一遍。这个设计彻底消除了对齐期间的阻塞Checkpoint 的耗时几乎只跟状态大小和存储带宽有关跟反压基本解耦。对于那种反压长期存在、倾斜严重的作业非对齐几乎是救命稻草我遇到过几个从两分钟降到五秒的案例。代价也很明确快照体积变大。状态本身 在途数据在途数据多的时候可能翻倍。恢复时间变长。除了要读状态还要把那些在途数据重新喂进流程走一遍。临时文件的管理成本。这些在途数据会先落到 TaskManager 的本地磁盘磁盘空间不够就直接失败。配置上要注意这个特性需要显式打开并且和状态后端有关。用 RocksDB 后端配合增量快照是最常见的组合。如果你打算开建议先在预发环境观察两周重点盯两个指标Checkpoint 大小有没有暴涨、恢复耗时有没有超出可接受范围。模式Checkpoint 速度快照体积恢复时间数据语义适用场景对齐 EXACTLY_ONCE受反压影响大正常正常精确一次反压可控、状态较小非对齐 EXACTLY_ONCE快且平稳偏大偏长精确一次反压大、倾斜严重对齐 AT_LEAST_ONCE快正常正常至少一次下游能去重非对齐 AT_LEAST_ONCE最快偏大偏长至少一次极端吞吐场景提示非对齐 Checkpoint 不是银弹。如果根因是外部系统写入慢、算子里有同步阻塞调用开非对齐只是把问题藏起来反压还是会从别的地方冒出来。3. Checkpoint 的存储与参数把能恢复调成恢复得快参数这一块很多人的做法是抄一份配置过来跑起来没报错就不管了。但 Checkpoint 参数之间是有内在联动的配错了不会立刻报错只会在某次故障恢复时集中爆发。我按重要性排一下顺序来讲。3.1 存储位置三种落地方式的取舍Flink 的 Checkpoint 数据最终要落到某个地方常见三种JobManager 堆内存。这是默认配置适合本地开发和状态极小默认单次不超过 5MB的场景。生产环境基本不能用原因有两个一是 JobManager 内存本来就紧张二是 JobManager 一旦重启快照全没了容错无从谈起。我见过有人在测试环境跑得好好的上了生产一重启作业就发现没得恢复就是因为没意识到这个默认值。本地文件系统。单机部署可以用分布式环境不行——TaskManager 挂了本地盘上的数据也就访问不到了。分布式文件系统或对象存储。这是生产标配。选型时主要看两件事写入带宽是否够决定 Checkpoint 耗时和小文件操作是否友好决定元数据操作的延迟。对象存储在大流量写入时偶尔会出现长尾延迟如果你的 Checkpoint 超时阈值设得很紧容易误判为失败可以把超时放宽一些。配置上要设两个目录一个是 Checkpoint 目录一个是 Savepoint 目录建议分开原因在第 4 节讲。state.checkpoints.dir: hdfs:///flink/checkpoints state.savepoints.dir: hdfs:///flink/savepoints execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION最后这行RETAIN_ON_CANCELLATION特别关键。默认值是DELETE_ON_CANCELLATION意思是作业被取消时它的 Checkpoint 会被一并删掉。你要是手滑点了取消或者升级时先停再启快照就没了只能从头消费。这个配置我建议所有生产作业都显式写上出事的时候能救命。3.2 间隔、超时、最小暂停三个参数必须一起看这三个参数的关系可以用一句话概括间隔决定多久拍一次超时决定拍一次最多等多久最小暂停决定两次之间至少歇多久。execution.checkpointing.interval: 30s execution.checkpointing.timeout: 10min execution.checkpointing.min-pause: 10s execution.checkpointing.max-concurrent-checkpoints: 1先看间隔该怎么定。这个值本质是在恢复代价和运行开销之间做选择。间隔 10 秒故障时最多丢 10 秒的数据但每秒都在做快照对存储和网络的持续压力大。间隔 5 分钟开销小但故障时可能要重放 5 分钟的数据如果这 5 分钟的数据量是几十 GB恢复期间的反压会非常恐怖。我的经验判断标准是看单次 Checkpoint 耗时占间隔的比例。如果一次 Checkpoint 要 20 秒间隔设 30 秒那意味着有三分之二的时间都在做快照这个作业基本没有余力处理业务高峰。健康的状态是单次耗时不超过间隔的三分之一最好在五分之一以内。再看超时。默认 10 分钟对大多数场景太宽松了——一次 Checkpoint 卡了 9 分钟还没失败中间所有依赖 Checkpoint 的动作比如两阶段提交的提交阶段都被挂住。我一般把它压到interval的 3 到 5 倍比如间隔 30 秒、超时 2 到 3 分钟。这个值的下限要高于正常情况下最慢一次 Checkpoint 的耗时的两倍否则容易出现误判。判断方法很简单翻一下历史监控里的 Checkpoint 耗时分布取 P99 再乘 2。最小暂停这个参数容易被忽视它控制两次 Checkpoint 之间至少间隔多久。设成 0 意味着上一次刚结束立刻开始下一次如果interval已经过了在状态大、快照慢的作业里会让 TaskManager 几乎没有喘息时间。设个 10 到 30 秒能明显降低 CPU 和 IO 的持续压力代价是实际间隔会比配置的更长。这里要清楚一个事实interval是最短间隔不是精确周期真正的两次间隔是max(interval, 上次耗时 min-pause)。max-concurrent-checkpoints一般保持 1除非你的状态特别大、单次特别慢才考虑设 2。设大了会让多个快照同时抢带宽反而都变慢还会增加存储上的并发写入压力。3.3 可容忍失败次数与保留数量execution.checkpointing.tolerable-failed-checkpoints默认是 0意思是只要有一次 Checkpoint 失败作业就直接挂掉重启。这个默认值在状态大、存储偶尔抖动的环境里太激进了一次网络抖动就能让作业重启重启又要重放数据得不偿失。设成 2 到 3 是比较常见的做法给系统一点容错空间。但要清楚一点这个参数只是延后了失败不是消除了失败。如果 Checkpoint 连续失败达到阈值作业还是会挂。所以它适合应对偶发失败不适合掩盖持续失败。持续失败必须去查根因通常是反压、外部系统超时或者存储权限问题。保留数量execution.checkpointing.num-retained决定磁盘上留几份历史快照。默认 1意味着新一轮成功后会删掉旧的。保留多份的好处是万一最新的一份损坏了还能回退到上一份。代价是存储占用成倍增加。我的做法是保留 2 到 3 份配合存储的生命周期策略成本可控。4. Savepoint 是运维坐标系不是手动版 Checkpoint很多人把 Savepoint 理解成我想什么时候拍就什么时候拍一次 Checkpoint这个理解只对了一半。两者在机制上有本质差别用错了会踩很深的坑。最核心的差别在于格式的兼容性承诺。Checkpoint 的格式是内部实现细节Flink 不保证跨版本兼容升级版本后老的 Checkpoint 大概率读不出来。Savepoint 的格式是对外承诺稳定的设计目标就是这个版本拍下的下个版本能读。所以版本升级、作业改造、跨集群迁移这些场景必须用 Savepoint不能用 Checkpoint。第二个差别是全量还是增量。Checkpoint 用 RocksDB 增量快照时只传新增的 SST 文件依赖之前的快照文件存在。Savepoint 一定是全量的把所有状态完整写一遍。这也解释了为什么 Savepoint 通常比 Checkpoint 慢、体积大好几倍。第三个差别是触发方式和生命周期。Checkpoint 是框架按周期自动触发、自动清理的Savepoint 是人为触发、默认永久保留、需要手动删的。这个差别看着小但直接决定了运维习惯——Savepoint 会越积越多磁盘会被慢慢吃满必须有清理机制。4.1 触发、停止、恢复的三段式流程标准的运维动作是三步每一步都有讲究。第一步触发 Savepoint。有两种方式一种是作业继续跑同时拍一张另一种是拍完就停。后者更常用因为升级场景本来就是要把作业停掉的。# 方式一只触发不停作业 flink savepoint jobId hdfs:///flink/savepoints # 方式二触发并停止推荐用于升级 flink stop -p hdfs:///flink/savepoints jobId # 如果是 YARN 上的会话模式还需要指定应用 ID flink stop -p hdfs:///flink/savepoints -yid appId jobIdflink stop比先 savepoint 再 cancel要好因为它是一个原子动作——拍完快照立刻停中间不会有新数据进来导致快照和实际停止位置不一致。第二步确认 Savepoint 完成。别急着启新作业先确认命令返回的路径真实存在且非空。我踩过一次坑命令返回成功但实际写入因为权限问题只写了个空目录启新作业时报找不到元数据文件白等半小时。第三步从 Savepoint 恢复。flink run -s hdfs:///flink/savepoints/savepoint-xxxxx \ -c com.example.MainJob \ app.jar恢复时会做一件事叫状态映射把 Savepoint 里记录的每个算子状态按 UID 匹配到新作业里对应的算子。匹配上了就恢复匹配不上就丢弃并打警告。这一步是所有问题的源头。4.2 算子 UID 是硬约束不是可选项Flink 会给每个算子自动生成一个 ID但自动生成的 ID 依赖算子在拓扑里的位置和顺序。你只要在中间加了一个算子、删了一个算子、或者调了一下链式合并的策略后面所有算子的自动 ID 就全变了Savepoint 里的状态找不到归属被当成无人认领的状态丢掉。表现出来就是作业能起来不报错但状态是空的所有计数从零开始所有去重逻辑失效。这种问题最阴险的地方在于它不报错你可能上线一周后才发现数据对不上。所以规矩只有一条所有有状态的算子必须显式设置 UID。DataStreamEvent stream env .addSource(new FlinkKafkaConsumer(...)) .uid(kafka-source) .name(kafka-source); stream.keyBy(Event::getUserId) .process(new DedupFunction()) .uid(dedup-process) .name(dedup-process);新版本 Flink 在恢复时如果发现算子没有 UID会直接拒绝恢复或者给出非常明确的告警。这个改动方向是对的逼着大家把这件事做对。我在项目里的做法是在代码规范里加一条硬性要求任何.process()、.keyBy()之后带状态的算子提交前必须 review 有没有.uid()。5. 状态结构变更与并行度调整从 Savepoint 恢复时真正会卡住的地方业务在变状态结构就不可能一成不变。给状态加个字段、改个类型、把两个状态合并成一个这些听起来都是小改动但在 Savepoint 恢复这一关每一个都可能让你卡住半天。5.1 加字段和改结构序列化器的兼容性边界状态是用序列化器写进去的读的时候也用同一个序列化器。你改了 POJO 的字段序列化器生成的 schema 就变了读老数据时可能报SerializationException也可能读到一堆 null。大致有这么几种情况处理方式不一样变更类型兼容性处理方式新增字段有默认值通常兼容确认序列化器能填默认值否则需自定义删除字段通常兼容老数据里的该字段被忽略字段类型变更不兼容需要写迁移逻辑或清状态POJO 改成 Scala Case Class不兼容换序列化器或清状态状态类型变更ValueState 改 ListState不兼容用StateProcessor做一次性迁移最稳妥的做法是给状态加一个版本号字段读出来先判断版本老版本的数据走迁移逻辑转成新结构再写回去。这样每次变更都是可控的不用赌序列化器的兼容性。如果状态不重要比如就是个缓存最省事的办法是配置忽略无人认领的状态让它自然丢弃接受一次冷启动。5.2 并行度调整key group 是那道绕不过去的墙前面提过 key group 的概念这里展开说。状态的分布是按 key group 切的key group 的总数由max parallelism决定默认 128。并行度调整时Flink 只是重新分配 key group 到新的并行实例上key group 总数不变所以状态不需要重新计算。这就带来两个结论第一并行度可以自由调整前提是max parallelism没变。从 4 扩到 8、从 8 缩到 2都没问题。这也解释了为什么默认 128 是个合理的值——它给了你足够的调整空间。第二max parallelism一旦显式设定就成了不可变的约束。如果老作业是用默认值 128 跑的新作业显式设成 256Savepoint 直接恢复失败。反过来也一样。所以要么两边都不设都用默认要么在第一次上线时就设好并且永不改动。还有一个坑即使max parallelism和并行度都对得上某些算子本身对并行度变化是敏感的。比如用了union之后做窗口聚合并行度变化会导致窗口的归属发生变化结果可能有偏差。这类算子要么在恢复前把状态清掉重新累积要么就保持并行度不变。5.3 一次完整的升级操作清单我把升级流程固化成了一个清单每次照着走出错概率低很多。确认当前作业健康。最近几次 Checkpoint 都成功耗时可接受。如果 Checkpoint 本来就在失败别急着升级先把问题解决。停掉上游写入或做好双写准备。如果是 Kafka可以先把消费组位点记下来便于回滚。执行flink stop触发 Savepoint记下返回的路径。验证 Savepoint。用flink run -s path在测试环境起一遍确认能正常恢复、状态非空、业务逻辑符合预期。这一步千万别省。启新作业观察三个指标恢复耗时、启动后前十分钟的反压、Checkpoint 是否恢复正常。保留旧 Savepoint 至少一周确认新作业没问题再手动清理。注意从 Savepoint 恢复时如果看到跳过 N 个无人认领的状态这类警告不要忽略。这说明有算子状态没对上号去核对 UID 和算子拓扑。6. 现场排查链路Checkpoint 反复超时和恢复慢的定位顺序前面讲的都是应该怎么做这一节讲出问题了怎么查。我把常见症状分成三类每类给一条排查路径。6.1 症状一Checkpoint 一直 IN_PROGRESS 不完成这是最常见的。界面上显示快照一直转圈最后超时失败。排查顺序应该是从最慢的地方倒着找。先看反压。界面上每个算子都有反压指标红色表示被下游压住了。找到第一个变红的算子问题就在它或者它下游。常见原因是外部系统写入慢数据库连接池不够、批量提交参数不合理、下游接口超时、算子里有同步阻塞调用、或者某个 key 的数据量远超其他。再看对齐时间。如果界面显示的状态是等待对齐说明有通道的 Barrier 迟迟不到本质还是数据倾斜或反压。这时候可以考虑开非对齐但更该做的是治倾斜。最后看存储写入。如果状态都已经快照完了卡在上传中那就是存储的问题——带宽不够、小文件太多、权限异常。看一下存储侧的监控或者换个大文件合并策略。还有一种隐蔽情况某个算子的snapshotState方法里做了耗时操作比如查了一次数据库每次快照都被拖住。这种只能靠代码 review 发现。6.2 症状二Checkpoint 频繁失败但作业本身在正常跑失败原因在异常信息里通常写得很清楚分类看CheckpointExpiredException超时。回到 6.1 的路径。IOException写存储失败权限、路径、磁盘满、对象存储限流。KafkaException提交位点失败通常是消费者和 broker 之间的网络或认证问题或者是位点提交超时设得太短。TimeoutException出现在某个 sink 上两阶段提交的 prepare 阶段卡住了检查 sink 的外部系统是不是在抖动。tolerable-failed-checkpoints在这里是缓冲垫。设了 2偶尔一次失败不会导致重启给你留出排查时间。但如果一天失败几十次必须查根因。6.3 症状三从 Savepoint 恢复特别慢恢复慢的原因通常是三个按发生概率排状态太大下载耗时长。几十 GB 的状态从远端拉到每个 TaskManager带宽是硬瓶颈。缓解办法是开本地恢复——把状态副本同时写一份到 TaskManager 的本地磁盘恢复时优先从本地读只有本地没有的才走远端。配置上需要打开开关并指定本地目录同时要注意本地盘的空间规划。state.backend.local-recovery: true taskmanager.state.local.root-dirs: /data/flink/local-state并行度不匹配导致大量数据重分布。恢复时如果并行度变了状态要在实例之间搬运网络开销很大。如果状态特别大尽量保持并行度不变。RocksDB 初始化慢。RocksDB 要打开所有的 SST 文件、重建索引状态文件多的时候这一步能花好几分钟。可以关注一下文件数量和单文件大小适当调整 compaction 策略。恢复慢还有一个经常被忽略的原因恢复后立刻开始消费同时在做状态加载两件事抢 IO。可以在启动脚本里加一点延迟让状态先加载完再开始拉数据实测能明显降低恢复初期的反压峰值。最后分享一个我在排查时常用的组合把 Checkpoint 的耗时、大小、失败原因三个指标放到同一张看板上按时间对齐。很多时候问题的拐点和某次发布、某次数据量突增是完全重合的看一眼就能定位到变更点。这个习惯帮我省下的时间比任何单点优化都多。
阅读完成 · 觉得有帮助？
