凌晨两点多被一条告警电话拽起来打开监控面板一看某个实时计算作业的checkpoint已经连续失败了七八次最近一次成功的 checkpoint 停留在半小时前。这个作业做的是订单维度的实时聚合下游还挂着看板和告警链路checkpoint 一直失败意味着状态没法持久化一旦进程挂掉就只能从很早的 savepoint 回滚中间的数据全部要重算。第一反应是去看反压结果反压指标确实有但只集中在少数几个 subtask 上其他 subtask 闲得发慌。顺着这条线索往下挖最后定位到是某个维度的 key 数据量暴涨把 keyBy 之后的分区彻底打歪了慢节点拖住了整条 checkpoint 的 barrier 对齐最终超时失败。这篇记录讲的就是这次问题的完整处理过程Flink 数据倾斜是怎么把 checkpoint 拖垮的、排查时应该看哪些指标、以及从临时止血到彻底治理的几种手段分别怎么用。适合正在跑实时作业、被 checkpoint 超时反复折磨的同学参考尤其是刚接手 Flink 作业、还不太会从 Web UI 里读出问题根因的朋友。里面涉及的参数配置、代码改法和踩过的坑都是我在实际生产环境里试过的可以直接对照着排查。1. 现象拆解checkpoint 失败到底在报什么1.1 三类高频报错指向的问题并不一样很多人看到 checkpoint 失败就笼统地认为是作业有问题实际上日志里的报错类型差别很大指向的根因完全不同。我这次遇到的作业日志里反复出现的是这么几类报错关键字字面含义常见根因Checkpoint expired before completing整个 checkpoint 在超时时间内没做完快照数据量大、上传慢、对齐耗时久Received barrier before.../Alignment timeoutbarrier 对齐超时反压、慢节点、数据倾斜Checkpoint was canceled because task was busy processing data任务忙于处理数据没空做快照CPU 打满、单点负载过高Checkpoint ... failed because of an exception快照过程抛异常状态后端 IO 异常、磁盘满、序列化问题这次作业的日志里Checkpoint expired before completing和Alignment timeout是交替出现的。这两个放在一起看基本就能把嫌疑锁定在barrier 在对齐阶段卡住了上而不是状态后端写不进去。这个判断很关键因为它决定了你接下来是去查存储还是查数据分布。我见过有同学一上来就怀疑 HDFS 慢折腾了半天存储配置最后发现是某个 subtask 数据量是别人的几十倍。1.2 Web UI 的 Checkpoint 页面是第一个要盯的地方Flink Web UI 里每个作业都有 Checkpoint 详情页点进最近一次失败的 checkpoint会看到一个耗时分解。这个分解比日志有用得多因为它把一次 checkpoint 拆成了几个阶段Start Delay从触发到真正开始做的时间如果这个值很大说明有任务在排队通常是反压或者资源不足。Alignment Durationbarrier 从 source 流到所有算子的对齐时间。这个值大基本就是数据分布不均或者反压。Sync Duration同步阶段主要是把状态刷到本地。Async Duration异步上传到远端存储的时间RocksDB 增量快照时这一项会比较明显。End to End Duration总耗时超过配置的 timeout 就失败。我这次的情况是Alignment Duration 占了总耗时的七成以上Sync 和 Async 都还算正常。这个比例一出来数据倾斜四个字基本就写在脸上了。你可以把最近十次 checkpoint 的这个分解拉出来对比如果 Alignment 一直是大头那方向就不会错。1.3 用 subtask 粒度的指标把嫌疑落到具体分区光知道对齐慢还不够得知道是哪个 subtask 慢。切到作业的 Metrics 页面看 keyBy 之后那个算子的 subtask 明细重点看这几个指标numRecordsIn / numRecordsOut每个 subtask 处理的记录数numBytesIn / numBytesOut数据量busyTimeMsPerSecond每秒忙碌时长接近 1000 说明打满backPressuredTimeMsPerSecond反压时长idleTimeMsPerSecond空闲时长我当时的截图里同一个算子有 16 个并行度其中 3 个 subtask 的 numRecordsIn 是其他 subtask 的 40 倍以上busyTime 一直贴着 1000ms/s几乎没有 idle而另外一边有七八个 subtask 的 idleTime 长期在 800ms/s 以上。这种旱的旱死、涝的涝死的分布是数据倾斜最典型的外观不需要再看别的指标就能下结论。2. 为什么倾斜的 key 会让 checkpoint 卡住2.1 barrier 对齐的本质是木桶效应要理解 checkpoint 为什么会被倾斜拖垮得先弄清楚 barrier 是怎么流动的。Flink 做一次 checkpoint 时会在 source 端插入一个 barrier这个 barrier 顺着数据流往下游走每个算子要等到所有输入 channel 的 barrier 都到齐才能做自己的快照然后把 barrier 继续往下游发。这个等所有 channel 到齐的过程就是对齐。问题就出在等这个字上。假设一个算子有 16 个并行度其中 3 个 subtask 因为数据量巨大处理每一条记录都要花很久那么它们处理 barrier 之前积压的数据就要花很长时间barrier 自然传得慢。而另外 13 个 subtask 早就处理完了barrier 也早就到了只能干等着。整个 checkpoint 的耗时取决于最慢的那条链路这就是典型的木桶效应。更糟的是这种等待是连锁的。慢 subtask 的上游会反压反压又会让上游的 barrier 推进更慢形成恶性循环。我在排查时看到某个 source 的 subtask 因为下游反压barrier 一直发不出去Start Delay 直接飙到十几秒。2.2 热点 key 的 state 膨胀会二次放大问题除了对齐慢倾斜还会从另一个角度打击 checkpointstate 体积。做聚合类的作业热点 key 的所有历史聚合结果都挂在同一个 subtask 的 state 里。其他 subtask 的 state 可能只有几百 KB这个热点 subtask 的 state 能到几个 GB。快照的时候它要序列化、要写盘、要上传时间自然比别人长一大截。这里有个容易被忽略的点如果你的状态后端是 RocksDBstate 大还会带来 compaction 的额外开销。写入过程中 RocksDB 在后台不停地 compaction本身就会抢 CPU 和磁盘 IO到 checkpoint 的时候同步把数据刷到 SST 文件如果碰上 compaction 正在跑耗时会被进一步拉长。我这次在监控里就看到热点 subtask 所在节点的磁盘写入延迟在 checkpoint 期间有明显尖刺。2.3 反压、buffer 和超时参数之间的连锁反应还有一层机制需要说清楚。Flink 的网络传输有 buffer 机制上下游之间通过 buffer 池传递数据。当慢 subtask 处理不过来它的输入 buffer 会被占满上游就发不出去反压信号一路往上传。反压状态下barrier 是插在数据流里的它也得排队前面堆积的数据没处理完barrier 就别想往前挪。这时候如果 checkpoint 的 timeout 设置得比较紧比如 10 分钟而积压的数据需要 15 分钟才能消化完那 checkpoint 必然超时失败。失败之后又会触发下一轮 checkpoint如果数据倾斜没有缓解新一轮还是同样的结局于是就出现了我看到的连续失败七八次。更麻烦的是checkpoint 失败本身也会带来开销比如清理临时文件、重试调度这些都会挤占本来就不宽裕的资源。所以处理这类问题光调大 timeout 只是把失败推迟根子还在数据分布上。这也是为什么我后面要分止血和根治两层来做。3. 排查链路一步步确认是倾斜而不是资源不足3.1 先把存储和网络的嫌疑排掉虽然从 Alignment Duration 占比高已经能大致判断但排查严谨一点总没坏处免得改完发现方向错了。第一步我会先排除外部因素看 checkpoint 存储的写入延迟和失败率。如果远端存储本身在抖动Async Duration 会异常高而不是 Alignment。我这次 Async 阶段稳定在几百毫秒存储侧没有明显异常。看节点之间的网络。跨机房的 shuffle 如果带宽打满barrier 传输也会慢。可以用监控看下节点间的网络吞吐和丢包。看是否有节点故障或者资源被其他作业抢占。如果某个 TaskManager 的 CPU 或内存被别的进程吃掉它上面的 subtask 自然会慢。这三项排查下来都没问题那就更坐实了是数据分布的问题。这里有个经验如果慢的 subtask 长期固定在同样的几个 slot 上大概率是数据倾斜如果慢的 subtask 每次都不一样、随机漂移那更可能是资源争抢或者 GC 问题。这个区分方法很实用我在好几套环境里验证过。3.2 采样找出是哪个 key 在作妖确认是倾斜之后下一步是找出热点 key。方法有好几种我一般会组合使用方法一在 keyBy 之前加计数器。用一个简单的RichMapFunction在 open 里注册一个 Counter 或者用 metric group然后每处理一条记录就按 key 的 hash 分桶计数。跑几分钟之后看哪个桶特别大再把桶里的 key 打出来。这个方法的好处是不影响主逻辑缺点是只能看个大概。方法二直接查源数据分布。如果上游是消息队列或者数据库直接写个查询按 key 分组 count 一下TopN 一目了然。我这次就是直接对源表跑了个group by dim_key order by count desc limit 20发现某个维度值占了总数据量的 60% 以上问题瞬间清晰。方法三在日志里采样。在 keyBy 之前按一定概率打印 key比如千分之一然后统计日志里的分布。这个方法比较粗糙但应急时能用。三种方法里我最推荐第二种代价最小、结论最直接。如果源数据不方便查再用第一种。3.3 区分倾斜发生在 source 端还是 keyBy 端这一步很多人会漏掉。数据倾斜可能发生在两个位置source 读取的时候就不均匀或者 source 均匀但 keyBy 之后被打歪了。判断方法很简单看 source 算子的各个 subtask 的 numRecordsOut。如果 source 端就已经严重不均那问题在上游分区比如消息队列的 partition 数据量不均、或者文件分片大小差异大。这种情况的解法是调整 source 并行度和分区数的匹配关系或者在 source 之后立刻加一个rebalance()或者rescale()把数据重新打散。如果 source 端是均匀的keyBy 之后才出现不均那就是典型的 key 分布问题需要用后面讲的两阶段聚合或者加盐来处理。我这次属于后者source 端 16 个 subtask 的读取量差异在 10% 以内keyBy 之后才炸开。4. 治理手段从临时止血到彻底根治4.1 先止血unaligned checkpoint 与超时参数作业还在线上跑下游还等着数据所以第一步是让 checkpoint 先能成功避免状态丢失的风险。我做了两件事开启非对齐 checkpoint。Flink 在 1.11 之后支持 unaligned checkpoint它的原理是不再等 barrier 对齐而是把 barrier 之前还在 buffer 里的数据也一起快照下来等恢复的时候再重放。这样就绕开了等最慢 channel的问题。配置项是execution.checkpointing.unaligned.enabled: true execution.checkpointing.unaligned.max-subtasks-per-channel-state: 4开启之后Alignment Duration 直接从几十秒降到接近零checkpoint 立刻能成功了。但要注意unaligned 会把 buffer 里的数据也写进 statestate 体积会变大恢复时间也会变长。所以它是个止血手段不是长期方案等业务侧治理完还是要考虑关掉。适当放宽超时和容忍失败次数。把 checkpoint timeout 从 10 分钟调到 15 分钟同时设置tolerable-failed-checkpoints为 3避免一次偶发失败就导致作业重启。这个配置要谨慎调太大等于掩盖问题我一般只作为过渡。需要提醒的是unaligned checkpoint 在某些场景下会和 exactly-once 的语义配合出问题尤其是和两阶段提交的 sink 一起用时要确认版本兼容性。上线前最好在预发环境跑一轮。4.2 两阶段聚合把热点 key 拆开算止血之后开始做根治。对于聚合类作业最通用的解法是两阶段聚合也叫 local-global 聚合。思路是先在本地加随机前缀把热点 key 打散做一次局部聚合然后再去掉前缀做全局聚合。这样同一个 key 的数据会被分到多个 subtask 上先算一遍压力就分散了。// 第一阶段加随机盐局部聚合 DataStreamTuple2String, Long localAgg sourceStream .map(new RichMapFunctionOrderEvent, Tuple2String, Long() { private Random random; Override public void open(Configuration parameters) { random new Random(); } Override public Tuple2String, Long map(OrderEvent event) { // 盐的个数建议是并行度的整数倍避免二次倾斜 int salt random.nextInt(16); return Tuple2.of(salt | event.getDimKey(), 1L); } }) .keyBy(t - t.f0) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .sum(1); // 第二阶段去掉盐全局聚合 DataStreamTuple2String, Long globalAgg localAgg .map(t - { String rawKey t.f0.substring(t.f0.indexOf(|) 1); return Tuple2.of(rawKey, t.f1); }) .keyBy(t - t.f0) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .sum(1);这里有几个细节必须注意。盐的个数不是随便定的一般取并行度的整数倍或者根据热点 key 的集中程度来定。如果盐太少打散效果不够盐太多第二阶段又会引入新的小文件或者小 state 问题。我这次并行度是 16热点 key 占了六成数据用了 16 个盐效果比较理想。两阶段聚合只适用于可结合associative的聚合函数比如 sum、count、max、min。像 avg 这种不能直接拆要先拆成 sum 和 count 两个指标分别聚合最后再相除。像 collect_list 这种有序聚合两阶段会改变顺序要谨慎。窗口的触发时机要对齐。两阶段如果都用同样的窗口第一阶段的结果会在窗口结束时才发出第二阶段再等一个窗口端到端延迟会翻倍。如果业务对延迟敏感可以考虑用 reduce 加定时器的方式或者用 Flink 的windowAll配合更短的窗口。4.3 自定义分区器与并行度调整的取舍除了两阶段聚合还有几种手段各有适用场景自定义 Partitioner。如果热点 key 是固定的那么几个可以在 keyBy 的时候用自定义分区器把热点 key 手动分配到不同的 subtask 上其他 key 走 hash。这种方式改动小但需要硬编码热点 key业务变化后要跟着改维护成本高。我一般只在前几种方案都不适用时才考虑。调整并行度。如果热点 key 无法拆开比如业务要求同一个 key 必须在一个 subtask 上处理那就只能给这个算子单独提高并行度。Flink 支持在算子级别设置setParallelism但要注意上下游并行度不一致时的数据重分布开销。另外提高并行度也意味着更多的 slot 和资源要评估集群容量。拆分作业。把热点维度的处理和普通维度的处理拆成两个作业各自独立扩缩容。这个方案最重但隔离性最好。我见过一些团队把大促期间的爆款商品单独走一条链路就是这个思路。下面这张表是我对几种方案的对比供参考方案改动成本效果适用场景两阶段聚合中需要改代码好能彻底打散可结合聚合、热点 key 多加盐自定义 key中好同上但更灵活自定义分区器低一般只能处理已知热点热点 key 固定且少提高并行度低有限治标资源充足、倾斜不严重拆分作业高最好热点业务需要独立保障4.4 状态后端与资源侧的配合优化数据分布治理完之后状态侧也要跟着调一调不然 checkpoint 还是快不起来。我做了这几项开启 RocksDB 增量 checkpoint。对于 state 比较大的作业全量快照每次都要把整个 state 传一遍代价很高。增量快照只传变化的 SST 文件能显著降低 Async Duration。配置是state.backend.incremental: true。开启后我观察到 checkpoint 的 Async 阶段耗时下降了大约六成。设置 state TTL。如果聚合结果只需要保留一段时间比如最近 7 天一定要配 TTL否则 state 会无限增长。TTL 的配置要注意清理策略onReadAndWrite会在读写时都检查过期比较稳妥但有一点性能开销。调整 RocksDB 的写缓冲和 compaction 参数。默认配置在小 state 下够用state 大了之后要适当调大write_buffer_size和max_write_buffer_number减少 compaction 的频率。这些参数没有万能值要根据 state 大小和磁盘性能实测。给慢节点更多内存。倾斜的 subtask 内存压力大容易触发频繁 GC。可以给 TaskManager 加内存或者调整taskmanager.memory.managed.fraction让 RocksDB 有更多可用内存。5. 上线验证与回退预案5.1 怎么确认问题真的解决了改完之后不能只看checkpoint 成功了就完事要确认几个指标都回到正常区间Alignment Duration应该稳定在秒级以内。如果还是几十秒说明打散不彻底。各 subtask 的 numRecordsIn最大最小值之间的差距应该收敛到合理范围比如 3 倍以内。差距还是很大说明盐的分布或者分区策略有问题。Checkpoint 的 state size总量可能因为 unaligned 或者盐的引入略有增加但单个 subtask 的 state 应该更均匀。端到端延迟两阶段聚合会引入额外延迟要确认在业务可接受范围内。反压指标busyTime 和 backPressuredTime 应该明显下降。我这次改完之后观察了两个小时Alignment Duration 从平均 40 多秒降到了 2 秒左右最慢 subtask 和最快 subtask 的处理量差距从 40 倍降到了 2 倍以内checkpoint 连续成功state 总量还比之前小了 15%因为热点 subtask 的 state 被拆散了RocksDB 的 compaction 也更均匀。5.2 几个容易回退的坑最后说说我在验证阶段踩过的几个坑都是改完之后才暴露出来的第一两阶段聚合的窗口语义要对齐。我一开始第一阶段用处理时间窗口第二阶段用事件时间窗口结果数据对不上。后来统一改成事件时间并且确认 watermark 能正常推进才恢复正常。两阶段的窗口类型和时间语义必须一致否则结果会错乱。第二unaligned checkpoint 和增量快照同时开启时恢复时间可能变长。因为 unaligned 会把 buffer 数据也写进 state恢复时要重放。如果作业对恢复时间敏感比如要求 5 分钟内拉起要评估这个影响必要时在倾斜治理完成后关掉 unaligned。第三加盐之后下游的 key 语义变了。如果有下游算子或者外部系统依赖原来的 key 格式要确保第二阶段已经还原。我在测试环境就遇到过下游解析 key 失败的问题原因是测试时只跑到了第一阶段。第四并行度调整后要重新评估资源。提高并行度会增加 slot 需求如果集群资源紧张可能导致作业排队或者频繁重启。调整之前先看集群的可用资源别把一个 checkpoint 问题换成一个资源问题。我自己在实际处理这类问题的体会是数据倾斜导致 checkpoint 失败表面上是个存储或者超时问题根子上还是数据分布和资源分配的问题。unaligned checkpoint 和调大 timeout 能快速止血但真正要稳定还得回到业务数据本身把热点 key 拆开、把 state 控制住、把资源配均匀。这套排查思路我在后面几次类似问题里复用基本都能在一两个小时内定位到根因比一开始漫无目的地翻日志高效得多。