1. Watermark 在流处理里到底扮演什么角色先说结论Watermark 不是迟到数据的开关而是一句向前推进的承诺——它告诉下游时间戳早于我的事件理论上不会再来或者来了我也不等。这句话说出口下游才敢关窗口、才敢算结果、才敢把状态清掉。我见过太多人把 watermark 当成延迟参数配了个forBoundedOutOfOrderness(Duration.ofSeconds(5))就以为万事大吉结果窗口不触发、数据算重、状态只涨不跌最后半夜爬起来看日志。这篇内容围绕 Flink 的WatermarkGenerator展开从接口定义、内置实现、装配链路一直聊到实际调参时踩过的坑。适合已经能跑通基本窗口作业、但还没搞清楚我这个窗口为什么就是不关的同学也适合想读 Flink 源码但被TimestampsAndWatermarksOperator那一坨绕晕的人。我不打算复述官方文档只讲源码里那些真正决定行为的东西。1.1 事件时间这条线其实只有两个角色把事件时间拆开看整条链路上只有两个角色在干活一个是时间戳提取器负责从每条记录里挖出一个long类型的毫秒值另一个就是WatermarkGenerator负责根据这些时间戳决定水位线推到哪。前者决定数据的时间坐标后者决定计算的推进进度。窗口、定时器、状态清理、双流 Join 的对齐全都建立在这两者的产出之上。有个容易混淆的点时间戳提取器只影响这条记录属于哪个窗口完全不影响什么时候关窗口。我调试过一次事故负责的同学把assignTimestampsAndWatermarks写在keyBy之后导致下游一个算子拿到的水位线和上游完全脱节——窗口在另一个并行度上算的水位线却是本地算的结果就是局部窗口早就该触发了却一直卡着。DataStreamEvent stream env.addSource(source) .assignTimestampsAndWatermarks( WatermarkStrategy .EventforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((e, t) - e.getEventTime()) );这段代码看起来人畜无害但它同时做了三件事注册了策略、创建了生成器、把生成器和下游算子的处理循环挂上了钩。后面几节就是把这三件事逐个拆开。1.2 源码里 Watermark 就是一个带时间戳的特殊记录去翻org.apache.flink.streaming.api.watermark.Watermark类本身简单到有点意外一个long timestamp字段一个getTimestamp()以及一个把水位线打印成Watermark ts的toString()。它没有任何是否对齐是否空闲之类的元数据这些语义全部由算子和阀门valve在运行时维护。public final class Watermark extends StreamElement { public static final Watermark MAX_WATERMARK new Watermark(Long.MAX_VALUE); private final long timestamp; public long getTimestamp() { return timestamp; } }MAX_WATERMARK这个常量在作业正常关闭时会被发出用来把剩余的所有窗口一次性推到触发状态。所以如果你在日志里看到Watermark 9223372036854775807别慌那是收尾信号不是数据异常。它也会走网络传输作为StreamElement的一种在算子之间流动。这意味着水位线是有真实传输成本的每发一个 watermark下游就多一次解析和一次对齐计算。这也是为什么默认的周期是 200ms 而不是 1ms——发得太勤算子间的消息量会明显上去。1.3 单调不减是硬约束不是建议Flink 对 watermark 有一条铁律只能前进不能后退。如果某个算子收到一个比自己当前水位线还小的 watermark它会直接丢弃并且在 DEBUG 级别打一条提示。很多窗口死活不触发的问题根子就在这里上游某条并行链路一直产出一个很小的时间戳把整条链路的水位线拽在原地。往下游传递时多输入算子比如keyBy之后经过connect或 union 的那个会对所有输入通道的 watermark取最小值。这个逻辑在StatusWatermarkValve里后面单独讲。记住一句话你的作业水位线永远等于最慢那个分区的进度。这个特性让水位线天然具备反压式的保守性但也让个别分区数据稀疏变成了一场灾难。2. WatermarkGenerator 接口的三把钥匙WatermarkGenerator是整个机制的唯一扩展点接口小到只有两个方法但它被调用的时机和频率差异极大这才是踩坑的重灾区。public interface WatermarkGeneratorT { void onEvent(T event, long eventTimestamp, WatermarkOutput output); void onPeriodicEmit(WatermarkOutput output); }onEvent每来一条记录就调一次onPeriodicEmit由处理时间定时器周期性调用默认周期由PipelineOptions.AUTO_WATERMARK_INTERVAL默认 200ms控制。这两者的分工决定了你写自定义生成器时的整个思路。2.1 onEvent 是观察者通常不应该在里面发水位线onEvent的典型职责只有一个记录到目前为止见到的最大时间戳。官方内置的BoundedOutOfOrdernessWatermarkGenerator就是这么做的public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { maxTimestamp Math.max(maxTimestamp, eventTimestamp); }它完全没有碰output。原因在于如果每条记录都发一次水位线就等于放弃了批量确认的意义——乱序数据里一个大时间戳进来又一个小时间戳进来你会疯狂地向前试探、又不得不因为不能后退而白白浪费信号量。周期性发送给了一个时间窗口去吸收抖动这才是乱序场景下最经济的选择。只有一种情况该在onEvent里发你用的就是标点式语义即碰到某个特定标记就当作一次结束比如读到一条type END_OF_BATCH的记录。这种场景下必须在onEvent里发出因为周期性发送根本不知道批次边界在哪。2.2 onPeriodicEmit 才是有界乱序的主战场周期调用的好处是节奏可控、频率稳定坏处是如果你什么都没做它就什么都不会发生。BoundedOutOfOrdernessWatermarkGenerator的onPeriodicEmit长这样public void onPeriodicEmit(WatermarkOutput output) { output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1)); }maxTimestamp的初值不是 0而是Long.MIN_VALUE outOfOrdernessMillis 1。这个初值的意义很关键在没有数据进来之前让maxTimestamp - outOfOrdernessMillis - 1恰好等于Long.MIN_VALUE避免一上来就发出一个看起来正常的水位线把下游坑死。Long.MIN_VALUE意味着完全不知道时间下游的窗口定时器不会被触发这是安全的。那个-1也值得说一句。设maxTimestamp 100乱序容忍 10 秒算出来是100 - 10000 - 1。为什么不是- 10000因为窗口的触发条件是watermark window.maxTimestamp()。假设窗口宽度 10 秒[0, 10000)这个窗口的maxTimestamp是 9999。如果水位线发 9000这个窗口不会触发如果我们保守到发 8999那就是多等了一个毫秒。但如果少减这个 1语义上可能出现水位线等于恰好等于窗口上界这种边界情况-1让水位线严格小于还能接收的最晚时间戳给边界留了一道缝。这种一毫秒的语义严谨是我读源码时最容易被忽略、也最值得学的地方。2.3 标点式生成器的正确写法标点式punctuated不是另一种周期而是由数据驱动。写的时候要注意你必须在onEvent里发onPeriodicEmit留空否则两种语义混在一起会互相打架public class BatchEndWatermarkGeneratorT implements WatermarkGeneratorT { private long maxTimestamp Long.MIN_VALUE; Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { maxTimestamp Math.max(maxTimestamp, eventTimestamp); if (isEndOfBatch(event)) { output.emitWatermark(new Watermark(maxTimestamp)); } } Override public void onPeriodicEmit(WatermarkOutput output) { // 标点式生成器保持沉默 } }有个现实问题标点式依赖数据里的标记一旦上游在某些时刻完全不发数据比如夜里流量为零水位线就永远停在那里下游窗口全部挂起。我做过一次日志聚合的作业白天一切正常凌晨两点开始窗口不输出查了半天才发现那一天某个分区的数据源把标记写丢了。所以我的习惯是标点式 周期性兜底。在onPeriodicEmit里加一个距上次发送超过 N 秒就发一次当前最大水位的逻辑比纯粹依赖标记稳得多。3. 从 WatermarkStrategy 到你写的算子装配链路走一遍WatermarkStrategy在 1.11 之后成了统一入口它本质上是一个工厂 一堆装饰器的组合。搞懂它的装配过程才能明白我配了 withIdleness 为什么没生效这种问题。3.1 策略被 createWatermarkGenerator 落地WatermarkStrategy的核心是抽象方法createWatermarkGenerator(WatermarkGeneratorSupplier.Context context)。内置的forBoundedOutOfOrderness、forMonotonousTimestamps、noWatermarks都是它的静态实现。注意这个方法的调用时机每个数据源分区、每个算子链的初始阶段各调用一次不是在提交作业时全局调一次。所以生成器实例是 per-subtask 的字段之间互不共享别指望在里面放全局计数器。withIdleness是在策略层面加的一层包装它不改变生成器行为而是在TimestampsAndWatermarksOperator里配合处理时间定时器工作。具体做法是记录最近一次收到数据的时间超过 idleTimeout 就把这个WatermarkOutput标记为 idle下游的阀门检测到 idle 通道后就不再把它的水位线纳入最小值计算。这个机制很关键但它有前提下游必须是能感知 idle 的StatusWatermarkValve。如果中间隔了一些自定义的ProcessFunction自己转发 watermark那 idle 标记就会丢失。3.2 算子里那个处理循环做了什么TimestampsAndWatermarksOperator的processElement大意如下public void processElement(StreamRecordT element) throws Exception { final long newTimestamp userFunction.assignTimestamp(element.getValue()); element.setTimestamp(newTimestamp); output.collect(element); watermarkGenerator.onEvent(element.getValue(), newTimestamp, watermarkOutput); }看清楚顺序先往下游发记录再调用 onEvent。这意味着 watermark 永远落后于同批数据至少一条记录的处理时间。对于标点式生成器这个顺序尤其重要——标记记录本身已经发给下游了然后水位线才跟上下游看到的顺序是记录 → 水位线符合直觉。周期性发送则是靠ProcessingTimeService注册的定时回调在onProcessingTime里调用watermarkGenerator.onPeriodicEmit(watermarkOutput)同时在回调里检查 idle 状态。这一段是 1.11 之后被重构过的早期版本的TimestampsAndWatermarks把周期逻辑塞在AssignerWithPeriodicWatermarks里回调时机依赖StreamStatus和ProcessingTimeService的组合读起来更绕。3.3 StatusWatermarkValve多输入的取最小值逻辑这是最容易被忽视、但影响最大的一个类。它的使命是一个算子有多个输入通道时如何把多个通道的 watermark 合成一个。规则大致三条通道状态对输出的影响全部 active水位线正常推进取所有 active 通道的最小值向前推进部分通道被标记 idle忽略 idle 通道只在 active 通道间取最小值通道数量不匹配 / 通道已结束该通道不再参与计算剩余通道继续它在推进之前会做一次对齐检查只有当水位线大于当前已输出的水位线时才发出去。这就是只能前进不能后退的实现位置。有些同学在自定义ProcessFunction里手动ctx.output(watermark)如果顺序搞错发了旧值就会在这里被静默丢弃从现象上看就是下游水位不动但日志里几乎没有提示。4. 三个内置生成器的选型账本Flink 提供的开箱即用生成器不多但选错一个代价可能是几小时的排查时间。我把它们放在一起做个对照。4.1 BoundedOutOfOrderness 是默认答案但不是万能答案WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))这是绝大多数作业的起点。它的计算逻辑简单到极致但参数怎么定才是真功夫。Duration砍得太大窗口延迟高、状态在大窗口场景下堆积砍得太小迟到数据全被丢进allowedLateness分支甚至直接扔进 side output。我的经验做法是先开一个统计任务用ProcessFunction把每条记录的eventTime和到达processingTime做差落一个分位数出来。取 P99 的差值作为初始值再上下浮动 20% 实测两轮。绝对不要拍脑袋写 10 秒或者 1 分钟——不同数据源、不同网络链路的乱序分布差异非常大同一条 Kafka 主题在跨机房同步之后P99 可能翻好几倍。4.2 AscendingTimestamps 是零容忍的特例public class AscendingTimestampsWatermarkGeneratorT extends BoundedOutOfOrdernessWatermarkGeneratorT { public AscendingTimestampsWatermarkGenerator() { super(Duration.ofMillis(0)); } Override public void onPeriodicEmit(WatermarkOutput output) { // 不需要onEvent 已经发了 } Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { output.emitWatermark(new Watermark(eventTimestamp)); } }它是BoundedOutOfOrderness的子类只是把乱序容忍压到了 0并且把发送动作从周期挪到了每条记录。这意味着数据严格单调递增时它的延迟最低几乎做到了数据到哪窗口关到哪。但只要来一条乱序数据那条数据的时间戳就小于当前水位线对应的窗口可能已经触发过了数据会被判为迟到。我一般在两种场景用它一是预聚合之后的数据本身已经按时间有序二是自己的测试环境用来验证窗口逻辑是否正确避免被乱序干扰。生产上除非你非常确定上游有序否则不要贪那点延迟。4.3 自定义生成器的真正用武之地自定义不是炫技而是当业务语义无法用容忍多少秒乱序表达时的唯一出路。我遇到过的典型场景有三类。第一类是按数据量而非时间触发。比如每攒够一万条就推进一次水位线适合高频小数据包的上报场景避免周期性空转。实现上就是用onEvent里的计数器配合output.emitWatermark。第二类是多级水位线。有些系统里同时存在业务事件时间和设备上报时间两条时间线需要按业务事件推进但容忍设备时钟偏差。这时候生成器要自己维护两套maxTimestamp取更保守的那个输出。第三类是水位线的非线性推进。比如流式训练场景希望水位线在训练轮次结束时推进一大步其他时候几乎不动。这类需求只能靠标点式生成器配合自定义的轮次标记来实现。写自定义生成器有两个硬性注意点一是必须保证输出单调不减宁可少推也不要回退二是不要在生成器里持有大对象因为它会参与 checkpoint 快照的序列化一个不小心就是NotSerializableException或者快照体积暴涨。5. 排查实录水位线不推进、窗口不触发、状态只涨不跌这一节是我最想写的部分。前面都是原理这一节是现场。我按现象分类把排查链路完整走一遍。5.1 现象一某些 key 的窗口永远不触发第一次遇到这个问题是在一个按设备维度聚合的作业上。大部分设备正常出结果少数设备一直不输出。第一反应是数据问题捞了半天发现那几个 key 在对应时间段压根没有数据。根因是数据空洞这几个分区的数据流不活跃onPeriodicEmit虽然被周期性调用但maxTimestamp一直停在很早的值上周期性发出的水位线是个极小的值。多输入算子取最小值整条链路的水位线就被这几个不活跃分区拖住了。修复方式就是withIdlenessWatermarkStrategy .EventforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((e, t) - e.getEventTime()) .withIdleness(Duration.ofMinutes(1))idleTimeout的取值要和业务的数据间隔对齐。定得太短正常的短暂空档会被误判成 idle导致水位线冲得太快、把还在路上的数据判成迟到定得太长问题依旧存在。我一般的取值是该分区正常数据间隔的 3 到 5 倍。还有一个隐蔽的前提只有数据源算子上的 withIdleness 才有效。如果你在keyBy之后另起一段 assignTimestampsAndWatermarks那个位置的分区已经是被 shuffle 过的idle 标记传不到想要的位置。这个坑我踩过一次改了半天参数都没用最后发现是位置放错了。5.2 现象二水位线看起来在动但窗口触发得特别晚这个现象背后通常有两个原因。第一个是并行度不均衡下游取的是所有通道的最小值如果某个并行度处理慢比如数据倾斜它的处理时间落后水位线自然落后。表现是整体不慢但就是差那几秒。排查方法是打开 Flink UI 的 Watermark 面板或者在某些版本里看currentInputWatermark指标逐个 subtask 对比。如果发现某个 subtask 的水位明显低于其他的基本可以锁定是数据倾斜导致的处理滞后。第二个原因是水位线传播路径上被有意或无意地拦截了。比如你在中间插了一个ProcessFunction里面重写了processWatermark却忘了调用output.emitWatermark那水位线到这就断了。这种 bug 的可怕之处在于窗口依然会运作只是永远慢一个人为延迟而且日志里不会有任何报错。public void processWatermark(Watermark mark) throws Exception { // 正确处理更新本地状态后再转发 currentWatermark mark.getTimestamp(); output.emitWatermark(mark); }5.3 现象三窗口触发正常但 checkpoint 越来越大这是我在一个长窗口作业里遇到的。窗口本身设了 1 小时水位线配置了 30 秒乱序看着毫无问题。但 checkpoint 大小在稳定增长从几百 MB 涨到几个 GB。真正的原因是窗口迟迟不关闭导致状态堆积而窗口不关闭的根源是水位线推进得不够快。在低频数据源上forBoundedOutOfOrderness只会输出maxTimestamp - outOfOrderness - 1如果数据本身稀疏maxTimestamp推进很慢一小时窗口实际要等到数据时间戳最大处 一小时才关。解决办法有两个方向一是缩短窗口长度、配合增量聚合reduce或aggregate把状态体积降下来二是调整水位线策略在稀疏数据源上改用标点式或自定义的按处理时间兜底推进逻辑。第二个方案治本但需要你非常清楚提前关窗会不会导致丢数据。我的建议是先在测试环境跑满 24 小时对比两种方案的结果差异再决定上哪个。5.4 现象四标点式生成器在故障恢复后行为异常标点式生成器依赖数据里的标记来推进水位线。作业从 checkpoint 恢复时会从最后一次成功快照对应的位点重放数据。如果重放的范围里恰好没有标记记录水位线就停在恢复时的值上下游窗口集体沉默直到新的标记到达。这个问题的根本在于水位线本身会被 checkpoint 保存但它不保证会被重新生成。我在恢复之后的监控里加了一条告警水位线超过 5 分钟不动就报警。事实证明这条告警救了我至少三次。另一个细节是WatermarkOutput有一个markIdle()方法它是幂等的但不是自动可以反复调用的。在自定义生成器里频繁调用markIdle和markActive会造成状态抖动下游的阀门需要不断重新计算最小值性能上是有损耗的。所以要控制好频率别在一个高频的onEvent里做这件事。6. 我在调 Watermark 时会走的几个固定动作最后聊点偏实操的东西。这些年下来我基本上形成了几个固定动作每次新作业上线前都会过一遍。第一个动作是先量乱序分布再定参数。写个临时的ProcessFunction把eventTime和System.currentTimeMillis()的差值按分钟维度打到日志里跑一两个业务高峰。分位数出来之后参数就八九不离十了。这一步只用十分钟能省掉后面好几个小时的扯皮。第二个动作是在 UI 上盯 Watermark 指标。重点看两个值当前水位线距离最新数据时间戳的差值也就是实际延迟以及各个 subtask 之间的水位线差异。前者用来判断参数是否合理后者用来发现倾斜。第三个动作是给水位线加停滞告警。不管是周期性还是标点式只要水位线在 N 分钟内没有推进就必须报警。这个指标比任何业务告警都灵敏因为它最早暴露问题。第四个动作是在水位线策略的位置上多看一眼。assignTimestampsAndWatermarks应该紧跟在 source 之后尽可能早。位置越靠后被中间算子影响的可能性越大idle 标记、时间戳的原始语义都可能被破坏。我在 code review 时看到这个方法出现在keyBy之后会直接打回去改。第五个动作是用forMonotonousTimestamps做对照实验。当怀疑窗口不触发是乱序导致的临时换成单调策略跑一小段如果窗口立刻正常了那就是乱序容忍参数配小了或者上游真的乱序严重。这个对照实验比翻日志快得多注意只在小流量环境做。再说一个细节WatermarkStrategy.noWatermarks()在某些场景下是有用的比如纯处理时间的作业或者你打算在更下游自己接管水位线。但千万别把它当成先关掉再说的开关——没有水位线所有基于事件时间的窗口都不会触发这是必然后果不是 bug。我个人在实际操作中的体会是Watermark 这块的复杂度不在接口本身而在它和窗口、状态、并行度、checkpoint 这几件事的交汇处。接口两个方法看懂只要五分钟但要知道一个参数改动会在多久之后以什么形式反馈到你的指标上需要真正被坑过几次。我现在的习惯是每次调整水位线相关配置都会同步改一下监控面板的阈值——因为这个改动的影响往往滞后十几分钟才显现等肉眼看到的时候窗口可能已经积了一堆没关的状态了。