1. 从一个反直觉的现象说起并行度调大后窗口反而迟迟不输出第一次在生产环境里把 Flink 作业的并行度从 2 调到 8我遇到过一个很典型的现象数据明明一直在流入窗口计算逻辑也没改但下游就是迟迟收不到结果等得人心慌。当时第一反应是是不是数据倾斜了查了半天发现各 subtask 的吞吐都挺均匀问题根本不在数据分布上而在watermark 在多并行度下的传播机制。这个坑其实很多人都会踩因为它违背直觉单并行度下跑得好好的逻辑一旦并行度上去watermark 的推进就变成了一个木桶效应——整个作业的 watermark 进度取决于最慢的那个 subtask。你只要有一个 subtask 因为数据稀疏、空闲或者反压而没推进 watermark所有下游算子的窗口都得跟着等。这篇内容我想把这件事彻底讲透。适合已经写过 Flink 窗口作业、但对 watermark 在多并行度下到底怎么流动还停留在大概知道阶段的同学。我会从 watermark 的本质讲起拆解它在并行任务间的传播链路重点分析空闲分区idle partition、watermark 对齐、多输入算子取最小值这几个最容易出问题的地方最后给出一套可以直接抄作业的排查和调优方法。全程用大白话加实际配置不堆公式但该有的原理一个不落。先说结论方便你带着问题往下看并行度越高watermark 被拖后腿的概率越大因为参与取最小值的 subtask 变多了任何一个掉队都会拖住全局。理解了这句话后面所有的现象都能解释。2. Watermark 到底在并行任务间怎么流动2.1 先厘清一个前提watermark 是广播出去的很多人对 watermark 有个误解以为它是跟着数据一起点对点传的。实际上在 Flink 里watermark 是沿着数据流的分区通道广播的。上游一个 subtask 产生的 watermark会发给下游所有它连接的 subtask。这里有个关键点上游的每个 subtask 是独立产生 watermark 的。假设上游并行度是 4那么就有 4 个 subtask 各自根据自己处理到的数据推进自己的 watermark。下游某个 subtask 如果同时接收这 4 个上游 subtask 的数据它就会收到 4 路 watermark。那下游怎么决定自己该用哪个 watermark 呢答案是取最小值。这个逻辑非常关键我们单独拎出来讲。2.2 多输入算子为什么必须取最小值想象一个下游算子它有两个输入通道 A 和 B。A 通道的 watermark 已经推进到 100 秒B 通道才到 80 秒。这时候算子能不能认为100 秒之前的数据都到齐了不能。因为 B 通道 80 到 100 秒之间的数据可能还在路上。如果算子贸然按 100 秒触发窗口B 通道那些本该属于这个窗口的数据就会被漏掉结果就是数据丢失或计算错误。所以 Flink 的规则很朴素一个算子实例的当前 watermark 它所有输入通道 watermark 的最小值。这个最小值就是整个并行作业 watermark 推进的瓶颈所在。用生活化的类比一个团队要宣布所有人都到齐了可以开会那必须等最晚到的那个人。哪怕 99 个人早就到了只要有 1 个人没来会就开不了。并行度越高这个团队越大出现迟到者的概率自然越高。2.3 一条完整的传播链路把上面的点串起来一条 watermark 从产生到触发窗口的完整链路是这样的源头 subtask 产生 watermark基于事件时间戳和配置的乱序容忍度out-of-orderness每个源 subtask 独立推进。中间算子接收多路 watermark对每个输入通道收到的 watermark 取最小值作为自己的当前 watermark。继续向下游广播中间算子把自己的 watermark 发给下游所有 subtask。窗口算子判断触发当 watermark 超过窗口结束时间减去允许延迟触发窗口计算。这条链路上任何一环的某个 subtask 卡住整条链路对应部分就卡住。这就是为什么并行度一高问题就冒出来。2.4 一个容易忽略的细节watermark 的传递是异步的watermark 在算子间传递走的是和普通数据一样的网络通道但它是异步的不保证和数据严格同步到达。这意味着即使上游已经发出了 watermark下游也可能因为网络缓冲、反压等原因延迟收到。这个特性带来的实际影响是你看到的 watermark 进度永远比上游实际产生的要滞后一点。在排查问题时不能只看上游日志里打印的 watermark 值要结合下游实际收到的来判断。我一般会在关键算子后面加一个自定义的 ProcessFunction把收到的 watermark 打出来这样能直观看到每一路输入的真实进度。3. 空闲分区并行度下最隐蔽的拖后腿元凶3.1 什么是空闲分区它为什么危险假设你的数据源是 Kafka有 8 个分区作业并行度也是 8。理想情况下每个 subtask 消费一个分区大家齐头并进。但现实往往是某些分区数据量很少甚至长时间没有数据。这时候那个消费稀疏分区的 subtask 就闲下来了。它没有新数据自然也就不会推进 watermark。而下游算子还在傻傻地等这一路 watermark于是整个作业的 watermark 就卡在这个空闲 subtask 上不动了。这就是空闲分区问题。它最坑的地方在于数据量大的分区跑得飞快但结果就是出不来因为被一个几乎没数据的分区拖住了。你去看监控CPU、吞吐都正常就是窗口不触发非常反直觉。3.2 用 withIdleness 让空闲源让路Flink 早就考虑到了这个问题提供了withIdleness配置。它的作用是如果一个源 subtask 在指定时间内没有收到任何数据就把它标记为空闲在计算下游 watermark 最小值时忽略它。配置方式是在 WatermarkStrategy 上调用WatermarkStrategy.EventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withIdleness(Duration.ofSeconds(30))这里的 30 秒就是空闲判定阈值。含义是某个源 subtask 连续 30 秒没数据就认为它空闲下游不再等它的 watermark。注意withIdleness的阈值不能设得太小。如果设成 1 秒而你的数据本身就是间歇性到达的比如每 5 秒一批那源会被频繁标记为空闲又恢复反而造成 watermark 抖动。一般建议设为正常数据间隔的 3 到 5 倍。3.3 空闲标记的恢复机制有个细节值得说清楚空闲状态不是永久的。一旦那个被标记为空闲的 subtask 重新收到数据它会立即恢复为活跃状态重新参与 watermark 最小值计算。这个恢复过程有个潜在风险如果空闲期间下游已经基于忽略它的 watermark 触发了窗口而恢复后这个 subtask 又送来了一批时间戳较早的数据这些数据就可能被判定为迟到数据。所以withIdleness配合允许延迟allowedLateness一起用才稳妥。我一般的组合是withIdleness设 30 秒窗口allowedLateness设 1 分钟这样即使空闲恢复后有迟到数据也能被窗口重新计算吸收掉。3.4 不用 withIdleness 的替代方案有些场景下withIdleness不太适用比如你无法接受忽略某个源带来的潜在数据延迟。这时候可以考虑另一种思路在数据源侧做保底心跳。具体做法是让每个源 subtask 即使没有业务数据也定期产生一条特殊的心跳记录带当前时间戳专门用来推进 watermark。这条记录在后续算子中被过滤掉不参与业务计算。这个方案的好处是 watermark 推进完全可控不依赖 Flink 的空闲检测坏处是要改动数据源逻辑且心跳记录会占用一定带宽。我一般只在withIdleness效果不理想时才用这招。4. 多输入算子的 watermark 对齐与取最小值陷阱4.1 两条流 join 时watermark 是怎么对齐的做双流 join 的同学对这个问题应该不陌生。假设流 A 和流 B 做 interval joinA 的 watermark 跑到 100 秒B 才到 60 秒。这时候算子会怎么处理答案是算子会缓存 A 的数据等 B 的 watermark 追上来。因为 join 需要两条流在时间上对齐B 还没到的部分A 对应的数据就不能确定是否有关联记录只能先存着。这就带来一个直接后果如果 B 流长期落后A 流的数据会在状态里越积越多最终可能导致状态爆炸或内存溢出。并行度越高参与对齐的 subtask 越多这种积压越容易发生。4.2 取最小值在多输入下的具体表现我们用一个具体例子把取最小值讲透。假设一个算子有两个输入输入通道当前 watermark通道 A100 秒通道 B60 秒算子当前的 watermark min(100, 60) 60 秒。也就是说尽管 A 已经跑到 100 秒算子对外表现出的进度只有 60 秒。如果通道 B 一直不推进算子 watermark 就永远停在 60 秒A 的数据全部积压。这就是为什么多流作业里慢的那条流会拖垮整个作业。4.3 并行度放大这个问题的机制单并行度下只有一条 A 通道和一条 B 通道取最小值就是两个数比大小。但并行度提到 8 之后情况变成上游 A 有 8 个 subtask各自产生 watermark。上游 B 有 8 个 subtask各自产生 watermark。下游某个 subtask 可能同时接收 A 的多个 subtask 和 B 的多个 subtask 的数据。这时候它要取的是所有输入通道 watermark 的最小值。参与比较的通道数从 2 变成了可能十几个。通道越多出现最小值特别小的概率越大这就是并行度放大问题的数学本质。用概率的话说假设每个通道有 10% 的概率落后单并行度下两个通道都正常的概率是 0.9×0.90.81并行度 8 时只要有一个通道落后就出问题正常概率降到 0.9 的十几次方几乎必然出问题。4.4 缓解多输入 watermark 瓶颈的实操手段针对这个问题我总结了几条实用手段让慢流变快如果 B 流本身数据量小考虑给它单独提高并行度或者优化它的处理逻辑让它尽快追上 A。控制乱序容忍度forBoundedOutOfOrderness的参数不要设太大。设 30 秒意味着 watermark 天然滞后 30 秒两条流叠加就更慢。一般设成实际乱序程度的 1.5 倍即可。用 interval join 替代 window joininterval join 对 watermark 的要求相对宽松状态管理也更精细能减少积压。监控每一路 watermark这是排查的前提后面单独讲。5. 一套可复现的 watermark 卡顿排查链路5.1 第一步确认是不是 watermark 问题窗口不输出原因可能有很多数据没到、逻辑写错、状态后端异常、watermark 卡住。怎么快速定位我的做法是在窗口算子前插一个探针把每个输入通道收到的 watermark 打印出来。如果发现某一路 watermark 长时间不动基本就能锁定是 watermark 问题。public class WatermarkProbe extends ProcessFunctionEvent, Event { Override public void processElement(Event value, Context ctx, CollectorEvent out) { out.collect(value); } Override public void onTimer(long timestamp, OnTimerContext ctx, CollectorEvent out) { // 定时打印当前 watermark System.out.println(subtask getRuntimeContext().getIndexOfThisSubtask() watermark: ctx.timerService().currentWatermark()); } }配合一个定时器每隔几秒打印一次就能看到每个 subtask 的 watermark 进度。5.2 第二步定位是哪一路输入拖后腿探针打印出来后对比各路 watermark。如果发现某一路明显落后继续往上游追是源 subtask 本身没数据空闲分区→ 用withIdleness。是源有数据但处理慢反压→ 查反压监控优化处理逻辑。是上游某个算子卡住→ 继续往上游插探针逐级定位。这个逐级往上游追的过程本质上是在找 watermark 传播链路上的瓶颈点。不要跳步一级一级查否则容易误判。5.3 第三步区分空闲和反压两种卡顿这两种卡顿表现相似watermark 都不动但处理方式完全不同必须区分清楚。特征空闲导致卡顿反压导致卡顿该 subtask 的输入速率接近 0正常或偏高该 subtask 的 CPU低高或饱和下游反压指标无有解决方向withIdleness / 心跳优化算子 / 加资源判断方法很简单看这个 subtask 的输入速率和 CPU。输入几乎为零、CPU 也低那就是空闲输入正常但 CPU 打满、下游有反压那就是反压。5.4 第四步验证修复效果改完配置后怎么确认真的修好了我的验证标准是探针打印的各路 watermark 差距在合理范围内比如不超过乱序容忍度。窗口能按预期频率触发不再长时间不输出。迟到数据比例在可接受范围内通过 Flink 的迟到数据指标观察。三个都满足才算真正解决。只看窗口输出了就收工很容易漏掉迟到数据变多这种副作用。6. 并行度与 watermark 参数的调优经验6.1 并行度不是越高越好很多人觉得并行度越高吞吐越大其实在 watermark 场景下并行度存在一个收益递减甚至反转的临界点。并行度提高参与 watermark 取最小值的 subtask 变多卡顿概率上升同时状态分片变多checkpoint 开销也变大。我的经验是并行度优先匹配数据源的分区数。Kafka 8 个分区并行度就设 8让每个 subtask 消费一个分区避免多个 subtask 抢一个分区或者一个 subtask 消费多个分区带来的不均衡。超过分区数的并行度对 watermark 推进基本没有正面作用反而增加协调成本。6.2 乱序容忍度的取值逻辑forBoundedOutOfOrderness的参数直接决定 watermark 的滞后程度。设得越大容忍乱序能力越强但窗口触发越慢设得越小触发越快但迟到数据越多。取值方法统计实际数据中事件时间与到达时间的最大差值取这个差值的 1.5 到 2 倍。比如实测最大乱序是 8 秒那就设 12 到 16 秒。不要凭感觉拍脑袋设个 1 分钟那会让所有窗口都白白等 1 分钟。6.3 空闲检测阈值和乱序容忍度的配合这两个参数要一起考虑。空闲检测阈值如果小于乱序容忍度可能出现源刚被标记空闲watermark 就跳变的情况导致窗口提前触发、迟到数据增多。我的建议是空闲检测阈值 ≥ 乱序容忍度的 2 倍。比如乱序容忍度 10 秒空闲检测就设 20 秒以上。这样 watermark 的推进比较平滑不会因为空闲标记的切换而剧烈抖动。6.4 一个真实调优案例的参数变化分享一个我实际调过的作业参数变化供参考参数调优前调优后调整理由并行度168匹配 Kafka 分区数减少 watermark 协调开销乱序容忍度60 秒15 秒实测最大乱序 8 秒60 秒过于保守空闲检测未配置30 秒存在稀疏分区需忽略空闲源允许延迟01 分钟配合空闲恢复吸收迟到数据调整后窗口触发延迟从原来的经常几分钟不输出变成稳定在 20 秒内触发迟到数据比例控制在 1% 以下。7. 几个我踩过的坑和对应心得7.1 坑一以为调大并行度能解决卡顿早期遇到窗口不输出我的第一反应是是不是并行度不够加并行度。结果并行度从 4 加到 16问题反而更严重了。后来才明白卡顿的根因是某个 subtask 拖后腿加并行度只会增加更多可能拖后腿的 subtask。心得遇到 watermark 卡顿先定位瓶颈再决定是加并行度还是减并行度。方向搞反了越调越糟。7.2 坑二withIdleness 阈值设得太激进有一次为了让窗口快点输出把withIdleness设成了 5 秒。结果数据本身是每 10 秒一批的源被反复标记空闲又恢复watermark 忽上忽下窗口触发时间变得完全不可预测还产生了一堆迟到数据。心得空闲检测阈值一定要大于正常数据间隔。数据是间歇性到达的场景这个阈值要设得保守一些宁可慢一点也不要抖动。7.3 坑三忽略了下游算子的 watermark 传播延迟有次排查发现上游 watermark 明明在推进下游窗口就是不触发。查了很久才发现中间隔了一个算子那个算子的 watermark 传播有延迟导致下游看到的 watermark 比上游滞后很多。心得watermark 排查要端到端看不能只看源头。在关键节点都插上探针才能看清完整的传播链路。7.4 坑四多流 join 时只优化了一条流双流 join 场景我只优化了数据量大的 A 流觉得 B 流数据少不用管。结果 B 流虽然数据少但因为处理逻辑复杂watermark 推进反而比 A 流慢成了新的瓶颈。心得多流场景下瓶颈往往在你想不到的那条流上。两条流都要监控 watermark不能想当然。8. 写在最后的一点个人体会Watermark 这个东西单并行度下看起来很简单一旦并行度上去、输入变多就变成了一门协调的艺术。它的核心矛盾始终是那句话整个作业的进度取决于最慢的那一环。理解了这一点你就能解释绝大多数窗口不输出的诡异现象。我现在排查这类问题基本形成了一个固定套路先插探针看各路 watermark再区分空闲还是反压然后逐级往上游追最后用迟到数据指标验证修复效果。这套流程走下来八九不离十能定位到问题。最后再分享一个小习惯上线前先在测试环境把并行度调到和生产一致用带时间戳的模拟数据跑一遍观察 watermark 推进是否平滑。很多并行度相关的问题在测试环境就能提前暴露比在生产上救火强得多。