企业官网建设流程全解析

Flink 作业的 checkpoint 失败，十次里有六七次，最后都能追到同一个根子上——数据倾斜。这不是什么玄学结论，是我这几年在实时链路上反复处理问题攒出来的经验：报错五花八门，有超时、有被新检查点挤掉、有协调器直接放弃，但把监控面板上的 subtask 维度数据拉出来一看，往往是某几个并行子任务在拖后腿，其余的在旁边闲着看戏。这篇文章就围绕一次典型的“数据倾斜导致 checkpoint 失败”的问题处理过程展开，把现象怎么读、证据怎么找、代码怎么改、参数怎么算、坑在哪里，一条线讲透。适合正在被 checkpoint 超时困扰的 Flink 开发、负责实时链路值守的运维同学，也适合刚开始接触大状态作业、还没踩过这类坑的朋友。文中涉及的具体参数取值来自实践中的经验换算，不是通用标准，请结合自己的数据量级做二次核算。

1. 现象还原：checkpoint 为什么会被拖垮

1.1 先看清报错，别急着把 timeout 调大

线上第一次看到这个问题，症状很朴素：作业没挂，数据也还在流，但 checkpoint 历史列表里连续几次都是红色，紧接着作业重启，重启后从上一个成功的检查点恢复，恢复期间积压暴涨，然后再次失败，形成一个很难自己走出来的循环。这时候最容易犯的错，是打开配置文件把execution.checkpointing.timeout从 10 分钟改成 30 分钟，然后安慰自己“应该够了吧”。

事实上，checkpoint 相关的报错文本本身携带的信息量很大，值得逐字读一遍。下面这张表是我整理的常见报错与它的真实指向，看懂这张表，能省掉一半的瞎猜时间。

关键的分水岭在于：是所有 subtask 都慢，还是只有个别 subtask 慢。前者通常指向存储层、网络层、或者全局限流；后者基本可以直接锁定倾斜。这个判断在面板上只需要十几秒，但它决定了后面几小时的工作方向，所以我一般把它放在最前面做。

1.2 把一次 checkpoint 拆成五个阶段来理解

想搞清楚倾斜是怎么把 checkpoint 打死的，得先知道一次 checkpoint 到底发生了什么。我习惯把它拆成五段：协调器在 Source 端注入 barrier 并向下游广播；barrier 在网络中随数据一起流动，每个算子需要等所有输入通道的 barrier 都到齐才能做本地快照，这一步叫对齐；对齐完成后做同步部分的快照，通常很轻；然后进入异步阶段，把内存里的状态刷到本地文件，再上传到远端存储；最后各任务向协调器确认，协调器宣布完成。

这五段里，对齐阶段和异步阶段是倾斜最容易打击的地方，但打击方式完全不同。倾斜导致某个 subtask 处理不过来，它的输入缓冲里堆满了待处理数据，barrier 只能排在数据后面慢慢往前挪，对齐时间就被拉长到几分钟甚至十几分钟。而如果倾斜体现在状态上——某个 key 对应的状态特别大——那么对齐可能很快，但异步阶段刷盘和上传会非常慢，单次 checkpoint 的体积能比其他 subtask 大出一个数量级。

上面这段区分非常重要，因为处理手段完全不同。对齐慢要去治反压和流量分布，异步慢要去治状态和存储。我见过不止一次，有人明明是对齐慢，却跑去调 RocksDB 的写缓冲，折腾两天没效果，最后发现方向从一开始就错了。

1.3 一条完整的因果链，以及它为什么会自我强化

把单点现象串成链条，你会看到这个问题为什么难缠。热点 key 集中在少数几个 subtask 上，这些 subtask 的 CPU 打满、处理速率上不去，于是输入队列积压，向上游传导成反压；反压让 barrier 无法及时穿过缓冲区，对齐时间从几秒涨到几百秒；超过超时阈值后本次 checkpoint 作废，如果失败次数超过tolerable-failed-checkpoints的容忍值，作业直接重启；重启后从上一个成功的检查点恢复，恢复期间所有积压数据一次性涌进来，热点 subtask 的压力比之前更大，于是下一次 checkpoint 更容易失败。

这条链最麻烦的地方在于它有正反馈。第一次失败可能只是偶发，但只要触发一次重启，积压就会放大倾斜的影响。所以处理这类问题，光把这一次救回来没用，必须把倾斜本身削平，否则就是反复救火。我在处理时的一般原则是：先止血保证作业能跑，再治病把倾斜削掉，最后补监控防止复发。止血手段可以是临时提高超时、打开非对齐检查点、临时加大并行度；治病手段才是改聚合逻辑、打散热点、优化状态结构。

1.4 为什么带倾斜的作业对 checkpoint 格外敏感

有人会问，同样有倾斜，为什么有的作业跑得好好的，有的就天天挂。差别在状态规模和并行度上。小状态作业，即使对齐慢一点，异步阶段几毫秒就完事，整体还能压在超时线以内；而大状态作业，异步阶段本身就要几百秒，对齐再叠几百秒，随便就超了。另外并行度越高，barrier 需要对齐的通道越多，“木桶效应”越明显——只要有一个通道慢，整条链就等它。

还有一点常被忽略：checkpoint 的对齐是逐算子进行的，倾斜往往发生在某个特定算子上，比如聚合算子。但这个算子的慢会通过反压传导到它上游的所有算子，导致上游算子的对齐也变慢。所以看面板时会发现“一片红”，误以为是全局问题，其实源头只有一个算子。定位时要逆着反压方向往回找，找到那个既是反压源头、又存在数据分布不均的算子，那才是真正要动刀的地方。

2. 定位过程：把“倾斜”从猜测变成证据

2.1 面板三板斧：流量、忙碌度、反压

Flink 自带的面板在定位倾斜这件事上其实够用，前提是你会看。我一般按三步走。第一步看算子级别的numRecordsInPerSecond和numRecordsOutPerSecond，把视图切到按 subtask 展示，如果某几个 subtask 的输入速率是其他 subtask 的三五倍，嫌疑就很明确了。第二步看busyTimeMsPerSecond，这个指标表示该 subtask 每秒有多少毫秒真正在干活，热点 subtask 通常贴近 1000，冷门的可能只有几十。第三步看反压，backPressuredTimeMsPerSecond高说明下游顶住了，idleTimeMsPerSecond高说明这个 subtask 在等数据。

这三个指标要组合起来读。如果某个 subtask 忙碌度满、反压高、输出速率低，那它是瓶颈；如果某个 subtask 空闲度高、输入速率低，那它是被“饿”着的，说明上游分区不均。真正有价值的组合是：同一个算子内，部分 subtask 忙碌度接近 1000 且反压高，另一部分忙碌度很低甚至大量空闲——这就是教科书级别的数据倾斜特征，几乎不需要再怀疑。

这里有个实操细节值得说：面板默认展示的是算子级别的聚合值，一定要手动切换到 subtask 维度，否则聚合之后的数字会把倾斜抹平，看上去一切正常。我见过有人就是因为没切视图，对着“平均值很健康”的面板看了半天，得出“没有倾斜”的结论。另外采样窗口别太长，用最近 5 到 10 分钟的窗口最能反映当前状态，用 1 小时的窗口会被历史数据稀释。

2.2 读 checkpoint 明细里的时间分布

面板的 Checkpoint 详情页能看到每一次检查点的耗时拆解，包括对齐时长、同步时长、异步时长、以及总时长。这个页面是判断“问题出在反压还是状态”的关键证据来源。我通常会把最近十次失败或接近失败的记录翻一遍，看时间主要花在哪一段。

还有一个容易被忽略的指标是单次 checkpoint 的状态大小。如果某次检查点的大小突然从 20GB 涨到 200GB，那不是倾斜，那是状态泄漏或者逻辑变更，方向完全不同。我一般的习惯是把状态大小的曲线和 checkpoint 时长的曲线放在一起看，两条曲线同步上涨，说明状态增长是主因；只有时长涨、状态大小没动，才更可能是反压。

2.3 线程栈与火焰图，把卡点钉死

面板能告诉你“慢”，但往往说不出“卡在哪一行”。这时候需要抓线程栈。做法很简单，进到对应 TaskManager 的进程里，连续抓五到十次线程栈，间隔一秒，然后看哪些栈帧反复出现。注意不要只抓一次，单次采样的偶然性太大，连续采样才能看出稳定的热点。

常见的几类卡点我整理了一下。如果栈里大量出现 RocksDB 的迭代或压缩相关调用，说明状态读写是瓶颈，方向是状态结构和 RocksDB 参数；如果栈里出现大量等待网络写出的调用，说明下游反压，方向是下游处理能力或者分区不均；如果栈里出现 TimerService 相关的调用，说明定时器数量过多，这在窗口或者大量注册定时器的场景里很常见；如果栈里频繁出现业务代码里的集合遍历或者序列化，那可能就是单条记录太重。

火焰图是线程栈的连续版，更适合看比例。抓的时候同样要覆盖足够长的时间窗口，最好能覆盖一个完整的 checkpoint 周期，否则采到的可能只是平时的处理片段，看不到 checkpoint 期间的异常。这一点很多人会忽略，采了三十秒就下结论，结果那三十秒刚好是空闲期。

2.4 用数据说话：把 key 分布统计出来

到这一步，倾斜已经基本确认了，但还缺一个量化的东西：到底多斜。因为后面加盐、拆 key、调并行度，都需要知道热 key 的占比。我的做法是从源端抽样，离线统计。抽样不用太多，百万级记录足够看出分布形态。

下面这段是常用的统计思路，用 Java 在本地跑，也可以换成任何你顺手的语言：

// 从源端抽样一批记录，统计 key 的分布，输出 Top N 与倾斜度 Map<String, Long> counter = new HashMap<>(); for (String key : sampleKeys) { counter.merge(key, 1L, Long::sum); } long total = sampleKeys.size(); List<Map.Entry<String, Long>> top = counter.entrySet().stream() .sorted(Map.Entry.<String, Long>comparingByValue().reversed()) .limit(20) .collect(Collectors.toList()); for (Map.Entry<String, Long> e : top) { double ratio = e.getValue() * 100.0 / total; System.out.printf("key=%s count=%d ratio=%.2f%%%n", e.getKey(), e.getValue(), ratio); } // 关键指标：Top1 占比、Top10 占比、以及最热与平均的倍数 double top1Ratio = top.get(0).getValue() * 100.0 / total; double avgPerKey = total * 1.0 / counter.size(); double skewFactor = top.get(0).getValue() / avgPerKey; System.out.printf("top1=%.2f%% avgPerKey=%.1f skewFactor=%.1f%n", top1Ratio, avgPerKey, skewFactor);

跑完之后看两个数：Top1 占比和倾斜倍数。我的经验阈值是，Top1 占比超过 5%，或者最热 key 是平均值的二十倍以上，就必须处理了，否则并行度开到多少都没用，因为一个 key 只能落在一个 subtask 上。这个结论很反直觉但很重要：keyBy 之后，单个 key 的处理能力上限就是单个 subtask 的处理能力上限，加并行度对这个 key 没有任何帮助。明白这一点，你就知道为什么“加资源”解决不了倾斜。

3. 处理方案：代码层、状态层、参数层三路下手

3.1 代码层改造：两阶段聚合与热点打散

最有效的手段永远在代码里。核心思路是把“一次全局聚合”拆成“局部预聚合 + 打散 + 二次聚合”三段。第一段在本地做预聚合，把同一个 subtask 内的相同 key 先合并，减少向下游发送的数据量；第二段给 key 加随机盐，把原本挤在一个 subtask 上的热点 key 拆成若干份分散到多个 subtask；第三段按原始 key 再做一次汇总，输出最终结果。

下面是一个可参考的实现骨架，假设做的是按商户维度的实时汇总：

// 第一段：本地预聚合，不产生 shuffle，先把同 key 的记录在本地合并 DataStream<MerchantAgg> localAgg = source .map(new LocalPreAggregator()) // 内部用 HashMap 做短窗口累加 .name("local-pre-agg"); // 第二段：加盐打散，盐的个数与热点 key 的量级相关 int saltBuckets = 8; DataStream<MerchantAgg> salted = localAgg .map(new RichMapFunction<MerchantAgg, MerchantAgg>() { private transient Random random; @Override public void open(Configuration parameters) { random = new Random(); } @Override public MerchantAgg map(MerchantAgg value) { int salt = random.nextInt(saltBuckets); return value.withSaltedKey(value.getMerchantId() + "#" + salt); } }) .name("salt-split"); // 第三段：按加盐后的 key 聚合，此时热点已被拆成 saltBuckets 份 DataStream<MerchantAgg> partial = salted .keyBy(MerchantAgg::getSaltedKey) .window(TumblingProcessingTimeWindows.of(Time.seconds(10))) .reduce(new MergeAgg()) .name("salted-reduce"); // 第四段：去盐，按原始 key 做最终合并 DataStream<MerchantAgg> result = partial .map(MerchantAgg::removeSalt) .keyBy(MerchantAgg::getMerchantId) .window(TumblingProcessingTimeWindows.of(Time.seconds(10))) .reduce(new MergeAgg()) .name("final-reduce");

盐的个数怎么定，不能拍脑袋。我的算法是：先估算热点 key 的每秒记录数，再估算单个 subtask 每秒能处理的记录数，两者相除得到需要的份数，向上取整，再留一点余量。比如热点 key 每秒二十万条，单个 subtask 每秒能处理五万条，那么至少四份，取八份更从容。注意盐太多也有代价：状态条目数成倍增长，checkpoint 体积变大，第二段的聚合窗口如果设置不合理，还会引入额外延迟。

还有一个更轻量的做法，如果倾斜不严重，只是偶发，可以只在第一段做预聚合，不加盐。本地预聚合能把同一个 subtask 内相同 key 的记录合并，虽然不能解决“同一个 key 的流量全落一个 subtask”的问题，但能显著减少下游压力，很多场景下这就够了。我一般先上预聚合观察一两天，还不行再加盐，避免一上来就把结构改复杂。

3.2 并行度与分区策略：别指望 rebalance 能救场

关于并行度，有几个常见的误解需要澄清。第一个误解是“加并行度能解决倾斜”。前面说过，单个 key 只能落在一个 subtask，加并行度对这个 key 无效，只会让冷门的 subtask 更闲。第二个误解是“调用 rebalance 就能均匀”。rebalance 是轮询分发，它让下游每个 subtask 收到数量相同的记录，但如果记录本身大小差异极大，或者下游还要按 key 聚合，那 rebalance 之后数据又被重新聚回热点 subtask，问题原样复现。

真正有用的分区手段有两类。一类是自定义分区器，在keyBy的KeySelector后面接一个Partitioner，对已知的热点 key 做特殊处理，比如把热 key 按后缀拆到多个 subtask，冷 key 走常规哈希。另一类是热点旁路，用侧输出流把热点 key 单独抽出来，用独立的算子甚至独立的作业处理，主流程只处理长尾。旁路的好处是隔离彻底，坏处是运维复杂度上升，要额外维护一条链路，适合热点非常集中且长期存在的场景。

还有一个务实的选择是调整并行度但不改分区，前提是你确认算子的并行度本身设小了。比如某个聚合算子并行度是 8，但输入量级需要 32 才能跟上，那扩到 32 确实有效，因为瓶颈是整体吞吐而不是分布不均。判断方法是看面板上所有 subtask 是否都接近满载，如果都满载，那是资源不够；如果只有几个满载，那是倾斜。这两种情况的处理手段完全相反。

3.3 状态层：给大 key 减负

如果问题表现在异步阶段（对齐很快，刷盘很慢），那要动的是状态结构。我遇到过几类典型的状态问题。第一类是单个 key 的状态条目过多，比如用MapState存了某个大商户的全部历史明细，动辄几百万条，RocksDB 每次快照都要刷这几百万条。第二类是大对象状态，比如用ValueState存了一个几十兆的列表，序列化和反序列化都极慢。第三类是ListState无限增长，没有清理机制，跑得越久越大。

对应的处理手段也分几种。能用MapState就不用ListState，因为MapState支持按 key 读写单条，不用整体序列化；给状态加 TTL，尤其是明细类状态，超过业务需要的时间就自动清理；把超大 key 的存储拆开，比如按时间分片，查询时再合并；实在需要保留全量历史，那就把它移出 Flink 的托管状态，放到外部存储，算子只保留索引或者最近一段。

配置层面也要配套。RocksDB 的写缓冲和块缓存大小直接影响快照和读写的快慢，下面这组参数是我在中等规模作业上常用的基线，需要按状态总量重新核算：

这里要强调一点：开启增量检查点不是免费的。它依赖本地保留的 SST 文件，意味着 TaskManager 的本地磁盘需要留出足够空间，而且恢复的时候要依赖这些文件，一旦本地盘被清空或者节点重建，恢复会退化甚至变慢。所以启用增量检查点的同时，一定要规划本地盘容量和清理策略。

还有一个容易被忘掉的优化：状态大的作业恢复慢，恢复期间又会触发新的 checkpoint，形成新一轮竞争。这时候可以考虑调大execution.checkpointing.interval让首次 checkpoint 晚一点来，或者把失败容忍次数调大一些，给作业留出追赶积压的时间窗口。这是典型的“止血”操作，不是治本，但能避免作业在恢复期反复横跳。

3.4 参数层：把时间预算算清楚

checkpoint 相关参数之间有明确的约束关系，很多人只盯着 timeout 调，忽略了它们之间的联动。核心约束是：timeout必须小于interval + min-pause的某种组合下的实际可用时间，同时interval要大于“单次 checkpoint 耗时 +min-pause”，否则会出现检查点排队或者互相挤掉。

我一般的算法是这样：先测出单次 checkpoint 的稳定耗时 P95，比如 120 秒；再留 50% 到 100% 的余量，得到预算 180 到 240 秒；timeout设在这个预算上，比如 300 秒；min-pause设为单次耗时的 30% 左右，比如 60 秒；interval就等于timeout的一半左右加上 min-pause，取 240 秒。这样排下来，一个检查点从触发到完成有足够余量，同时两个检查点之间不会挤在一起。

# 参考配置片段，需按实际量级重新核算 execution.checkpointing.interval: 240s execution.checkpointing.timeout: 300s execution.checkpointing.min-pause: 60s execution.checkpointing.max-concurrent-checkpoints: 1 execution.checkpointing.tolerable-failed-checkpoints: 3 execution.checkpointing.mode: EXACTLY_ONCE state.backend: rocksdb state.backend.incremental: true

max-concurrent-checkpoints这个参数在倾斜场景下值得多说一句。设成 1 意味着上一次没完成就不触发下一次，好处是不会堆积，坏处是如果单次耗时超过 interval，实际检查点间隔会被拉长，恢复点变稀疏。设成大于 1 能让检查点更密集，但并发执行会加重负担，在已经反压的作业上往往适得其反。我的默认建议是 1，只有在确认瓶颈是存储带宽而非计算资源时，才考虑放宽。

至于非对齐检查点，它是这类问题的“止血带”。原理是不再要求 barrier 对齐，而是把对齐期间缓冲区里的数据也一并快照进去，这样慢的 subtask 不会拖住整条链。开启之后，对齐时间通常能直接降到接近零，checkpoint 立刻能成功。但代价要清楚：快照的数据量变大，文件数变多，恢复时需要重放更多数据，恢复时间可能反而变长；而且它只是绕过了问题，倾斜还在，吞吐还是上不去。所以我一般把它当作过渡手段——先打开让作业跑起来，同时并行做代码改造，改造完成后再评估是否关掉。

4. 踩坑记录与常见问题速查

4.1 四个真实踩过的坑

第一个坑：加盐加错了位置。最开始我把加盐放在窗口聚合之后，结果发现状态反而更大了，因为加盐后每个盐桶都保留了一份窗口状态，总条目数翻了几倍，checkpoint 体积不降反升。正确做法是把盐加在聚合之前的 keyBy 上，让打散作用在数据分发阶段，而不是状态存储阶段。这个顺序搞反，等于自己给自己挖坑。

第二个坑：只调 timeout 不治理。有一段时间我图省事，把 timeout 从 5 分钟调到 20 分钟，作业确实不重启了，但每次都跑到 18 分钟才完成，期间积压越来越严重，吞吐掉了一半。更糟的是，一旦某次超过 20 分钟，作业重启后要从 20 分钟前的点恢复，回追的数据量巨大，恢复时间比之前更长。调大超时只是把问题往后推，代价是恢复点变稀疏，风险反而变大。

第三个坑：非对齐检查点打开后没关注文件数。开了非对齐之后，一次检查点产生的文件数从几百个涨到几万个，远端存储的元数据操作成了新瓶颈，上传阶段反而变慢，而且恢复时要逐个打开这些小文件，恢复时间从三分钟涨到二十分钟。后来调整了文件合并策略，并且只在真正的流量高峰时段开启，情况才好转。

第四个坑：只改并行度不改分区，倾斜原样迁移。有一次把一个聚合算子从 16 并行度扩到 64，扩完之后看面板，倾斜比例一点没变，只是热点从一个 subtask 变成了四个 subtask，每个还是那么忙。原因是热点 key 还是那些，keyBy 的哈希结果决定了它们只会落在固定的几个 subtask 上。扩容解决的是整体吞吐，不解决单 key 热点，这个认知必须建立起来。

4.2 常见问题速查表

这张表里的最后一行值得单独说。有时候检查点没失败，但你会发现两次检查点之间的间隔在慢慢变长，这其实是作业在“隐性恶化”。原因是单次耗时已经逼近 interval，max-concurrent-checkpoints为 1 时，新的检查点只能等旧的完成。这种状态比直接失败更危险，因为不报错，没人注意，直到某天真的超时才发现。监控上要把“检查点间隔”也纳入观察，而不只是看成功失败。

4.3 监控与预防：让下一次倾斜早点暴露

处理完这一次之后，我把监控补了几项，后来确实提前发现过两次苗头。第一项是subtask 维度的处理速率差异比，取同一算子内最大值与中位数的比值，超过 3 就告警。这个指标比看总量敏感得多，总量正常但内部差异大，就是倾斜的前兆。第二项是对齐时长的 P99，超过阈值告警，因为对齐变长往往早于失败出现。第三项是状态大小的日环比，涨幅超过某个比例就去看是不是有 key 异常膨胀。第四项是RocksDB 的待压缩字节数，这个指标飙升往往预示着异步阶段要变慢。

预防方面，最有效的其实不是监控，而是上线前的压测。压测时要有意识地构造热点数据，比如让某一个 key 占总量的一成以上，看作业在这种分布下能不能扛住检查点。很多问题在均匀分布的测试数据下根本暴露不出来，一上生产就现原形。我自己现在的习惯是，凡是涉及 keyBy 聚合的作业，压测数据里一定掺入热点，这已经成了固定动作。

最后分享一个我觉得挺有用的小技巧：在聚合算子里加一个轻量的埋点，统计每个 subtask 处理的记录数和去重 key 数，定期打到日志或者指标系统里。这个埋点成本很低，但它能直接告诉你“哪个 subtask 在处理多少 key”，比事后通过各种间接指标推断要直观得多。我在几个关键作业上都加了这个，排查效率提升明显，尤其是那种倾斜只在特定时段出现的场景，回头看埋点数据，一眼就能定位到时间点和具体的 key。