做流处理这几年被问得最多的一类问题几乎都和 Flink 的 Watermark 有关。不是“Watermark 是什么”这种概念题而是“我明明设置了 WatermarkGenerator窗口为什么就是不触发”“水位线卡在某个值上一动不动”“并行度调大之后延迟突然涨了好几倍”。这些问题的共同点是只看官方文档里的那几行示例代码看不出答案必须翻到 WatermarkGenerator 的实现里看清楚它被谁调用、什么时候调用、调用之后那根水位线又经过了几层包装才流到下游。这篇内容就是把我自己读 Flink 源码的路径重新走一遍从 WatermarkGenerator 这个只有两个方法的接口切入一路跟到 TimestampsAndWatermarksOperator再到多输入场景下的水位线裁决逻辑中间穿插我自己写自定义生成器时踩过的坑和验证手段。适合已经能跑通 Flink 作业、但对事件时间和水位线心里没底的兄弟也适合想动手改 WatermarkGenerator 但不确定副作用的人。1. 先把问题摆到桌面上为什么非要有这根水位线1.1 从窗口触发的那个死结说起假设你在做一个按分钟统计的实时看板数据来自一堆客户端上报每条记录自带一个发生时间戳。业务上你要的是“按照数据真正发生的时间每 1 分钟聚一次”也就是事件时间语义。现在问题来了你要等到什么时候才敢把 10:00 到 10:01 这一分钟的窗口算出来并输出结果这里有个绕不过去的死结。数据在网络里是会乱序的10:00:59 的那条记录可能比 10:00:03 的那条晚到十几秒。如果你一看到时间戳越过 10:01:00 就把窗口结算了那后面姗姗来迟的记录就只能被丢掉或者重算。反过来如果你一直等等到“确定不会再有 10:01 之前的数据了”那这个“确定”本身就需要一个外部信号——流处理系统自己没法凭空知道未来还有没有数据。Watermark 就是这个外部信号的形式化表达。它是一条特殊的数据记录只带一个时间戳语义是“事件时间小于等于这个时间戳的数据我认为已经全部到齐了。”注意这里用的是“我认为”不是“我保证”这个措辞差别很关键后面讲迟到数据的时候还会回到这一点。所以整个机制的逻辑链条是数据自带时间戳 → 用一个生成策略持续地抬升水位线 → 算子收到水位线后把“已到期”的定时器比如窗口结束时间触发掉 → 结果输出。水位线的推进速度直接决定了你的结果延迟和结果准确性之间的平衡点而这个平衡点就是 WatermarkGenerator 要负责的事情。1.2 Watermark 在源码里到底长什么样很多人第一次翻到org.apache.flink.streaming.api.watermark.Watermark这个类的时候都会有点意外因为它实在是太简单了。它继承自StreamElement和StreamRecord、WatermarkStatus是平级的核心就只有一个long timestamp字段外加一个常量MAX_WATERMARK new Watermark(Long.MAX_VALUE)。这个设计其实挺说明问题的。水位线在运行时就是一条普通元素跟着数据记录一起在算子之间流动走的是同一套网络传输和反压机制。它不是某种控制平面的元数据也不是在某个中心节点维护的全局变量。这意味着两件事第一水位线会像数据一样被缓冲、被反压卡住如果一个算子处理慢它上游的水位线也会跟着堵第二水位线在跨算子传递时是有分区概念的这一点在后面讲并行度的时候会展开。另外一个容易被忽略的细节是Watermark类没有实现任何比较逻辑也没有单调性校验。也就是说框架层面并不会阻止你发出一个比上一条更小的水位线虽然那样做在语义上是错误并且会导致下游定时器状态混乱。这个约束是靠WatermarkOutput那一层的包装来兜底的而不是靠Watermark自己。理解这一点很重要你写自定义生成器时如果直接拿到WatermarkOutput就乱发很可能不会立刻报错而是过一段时间之后在某个窗口算子里看到莫名其妙的结果。1.3 三条语义保证和它们各自的代价把水位线的语义拆开实际上有三条相互关联的保证在起作用。第一条是单调性同一个输出通道上的水位线只能增不能减。这条保证让下游可以安全地用一个long变量记住“当前水位线”不用做任何比较。代价是如果一个分区长时间没有数据它的水位线就不动会把整个下游卡住——这直接催生了空闲检测idleness机制。第二条是全序到偏序的转换单个分区内的水位线是单调的但多个上游分区之间没有顺序关系。下游算子必须把它们合并成一个合并规则是取最小值。取最小值是为了保证“所有上游都说这个时间点之前的数据到齐了”才成立任何一路没到齐就不能推进。代价是只要有一路慢全局都慢这是水位线最容易出问题的地方。第三条是不保证不丢数据水位线只表达“我认为到齐了”不表达“我保证到齐了”。越过水位线之后到达的数据就是迟到数据默认会被丢弃。代价是你要么加大乱序容忍度换取准确性、牺牲延迟要么开启allowedLateness让窗口多留一会儿要么用侧输出把迟到数据接出来单独处理。这三条放在一起看Watermark 的整个设计取向就清楚了它不追求绝对正确它追求的是在有限延迟内给出一个可以量化的近似结果把“要多准”这个决策权交给使用者。而这个决策权的具体承载物就是接下来要拆的WatermarkGenerator。2. WatermarkGenerator 接口两个方法背后的调用时序2.1 onEvent 与 onPeriodicEmit 分别被谁调用打开org.apache.flink.api.common.eventtime.WatermarkGenerator你会看到一个特别克制的接口定义只有两个方法onEvent(T event, long eventTimestamp, WatermarkOutput output)和onPeriodicEmit(WatermarkOutput output)。没有 init没有 close没有获取当前水位线的 getter。这种极简设计是有意为之的接口越窄框架对实现的假设就越少。onEvent的调用时机很明确每处理一条数据就调一次在TimestampsAndWatermarksOperator.processElement里紧跟在时间戳提取之后。注意传入的eventTimestamp是已经由TimestampAssigner提取好的值不是你原始数据里的字段所以如果你在onEvent里再去解析一遍字段纯属浪费。onPeriodicEmit的调用时机就有讲究了。它是由算子内部基于处理时间注册的一个定时器周期性触发的间隔由pipeline.auto-watermark-interval控制默认 200 毫秒。这里的关键词是“处理时间”——也就是说即使你的数据流完全静止、一条数据都不来只要这个定时器还在跑onPeriodicEmit就会被调用。这解释了一个常见困惑“我的流里没数据了为什么下游还在收到水位线”答案是周期型生成器在空转。还有一个隐蔽的细节当auto-watermark-interval被设置为 0 的时候周期定时器根本不会被注册onPeriodicEmit永远不会被调用。这就是为什么纯打点型punctuated生成器必须把这个间隔设为 0否则就是白白浪费一次回调。2.2 周期型和打点型本质是同一接口的两种用法官方文档把水位线生成分成“周期型”和“打点型”两类但从源码角度看这两者并没有对应的两个接口它们只是同一个WatermarkGenerator接口的两种填充方式。周期型的做法是onEvent里只更新内部状态比如记录见过的最大时间戳真正的emitWatermark放在onPeriodicEmit里。好处是发水位线的频率和数据量解耦吞吐高的时候不会因为每条数据都发一次水位线而把网络打满。坏处是水位线的推进有最多一个 interval 的延迟精细控制做不到。打点型的做法反过来onEvent里判断某个条件比如遇到一个特殊标记事件、或者跨过了某个整数边界条件满足就立刻output.emitWatermark(...)onPeriodicEmit留空。好处是响应及时语义上可以和业务事件绑死。坏处是如果条件设计得不好可能出现长时间不发水位线导致下游窗口不触发或者发得太频繁造成额外开销。维度周期型打点型状态更新位置onEventonEvent水位线发射位置onPeriodicEmitonEvent发射频率决定因素处理时间间隔数据内容数据静止时是否推进是会空转否典型配置interval 默认 200msinterval 设为 0适合场景绝大多数无特殊标记的流有明确结束标记或分片边界的流实际项目里我基本都用周期型因为绝大多数数据源没有可靠的“分片结束”标记硬凑一个打点条件反而容易出错。只有在处理那种带明确批次结束标记的数据比如文件导入、MQ 里的批次完成消息时打点型才有明显优势。2.3 WatermarkOutput被包装了一层的那道闸门WatermarkOutput这个接口名字看着像输出端实际用起来更像一道闸门。它只有一个emitWatermark(Watermark watermark)方法和一个markIdle()新版是emitWatermarkStatus(WatermarkStatus)。为什么需要这层抽象因为生成器发出去的水位线不一定会原封不动地流到网络上去。在算子内部这个output可能是一个WatermarkOutputMultiplexer.Output也可能是一个带着空闲状态跟踪的包装。最典型的是WatermarkOutputMultiplexer的场景当一个算子的输出被链式chained连接到多个下游输入时生成器只发一次水位线多路复用器要把它记录下来然后对每一路下游分别计算“应该看到的水位线”再分别转发。这里有个很容易忽略的点WatermarkOutputMultiplexer在合并多路输出时取的也是最小值。也就是说如果你在一个ProcessFunction里做了分流两路下游的处理速度不一样慢的那一路会拖住快的那一路看到的水位线吗答案是会因为多路复用器是取所有 active 输出的最小值来算出“组合水位线”的。这是从正确性出发的保守设计实际表现就是分流之后整体水位线推进变慢。我在一个双流 join 的项目里被这个坑过一次排查了半天才意识到是其中一路下游反压导致水位线整体停滞。3. 沿源码往下走从一个接口到真正的水位线3.1 BoundedOutOfOrdernessWatermarks 到底做了什么几乎所有人第一个接触的实现都是BoundedOutOfOrdernessWatermarks也就是WatermarkStrategy.forBoundedOutOfOrderness(Duration)背后那个类。它的代码短得可怜但每一行都有讲究。public class BoundedOutOfOrdernessWatermarksT implements WatermarkGeneratorT { /** The maximum timestamp encountered so far in the stream. */ private long maxTimestamp; /** The maximum out-of-orderness that this watermark generator assumes. */ private final long outOfOrdernessMillis; public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) { checkNotNull(maxOutOfOrderness, maxOutOfOrderness); checkArgument(!maxOutOfOrderness.isNegative(), maxOutOfOrderness cannot be negative); this.outOfOrdernessMillis maxOutOfOrderness.toMillis(); // start so that our lowest watermark would be Long.MIN_VALUE. this.maxTimestamp Long.MIN_VALUE outOfOrdernessMillis 1; } Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { maxTimestamp Math.max(maxTimestamp, eventTimestamp); } Override public void onPeriodicEmit(WatermarkOutput output) { output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1)); } }先看构造函数里那个初始值Long.MIN_VALUE outOfOrdernessMillis 1。第一眼看上去很奇怪为什么要加个outOfOrdernessMillis 1原因是这一行同时防了两件事。一是防止下溢如果maxTimestamp从Long.MIN_VALUE开始onPeriodicEmit里算Long.MIN_VALUE - outOfOrdernessMillis - 1就会直接溢出成一个巨大的正数发出去一条Long.MAX_VALUE级别的水位线下游所有窗口瞬间全部触发。二是保证初始水位线恰好落在Long.MIN_VALUE上这样“还没收到任何数据”的状态不会误触发任何定时器。再看onPeriodicEmit里那个- 1。这个才是最容易看漏的地方。假设乱序容忍度是 0AscendingTimestampsWatermarks的情况那么发出的水位线就是maxTimestamp - 1而不是maxTimestamp。为什么要减这个 1因为 Watermark 的语义是“小于等于该时间戳的数据已经到齐”而窗口的触发条件是watermark window.maxTimestamp()其中window.maxTimestamp()对于时间窗口来说是endTime - 1。如果不减 1那么当水位线正好等于maxTimestamp时maxTimestamp本身那条数据对应的窗口边界可能还没到就会出现“水位线到了但窗口没触发”或者“窗口提前触发”的偏差。这个 1 是边界对齐的修正项不是拍脑袋。3.2 时间戳提取器和生成器的组装过程单独一个WatermarkGenerator是不能工作的它必须和一个TimestampAssigner一起打包成WatermarkStrategy。而这个打包动作发生在作业图构建阶段不是在运行时。WatermarkStrategy的实现类WatermarkStrategy.WithTimestampsWatermarks里有个createWatermarkGenerator方法它会调用WatermarkGeneratorSupplier.createWatermarkGenerator(context)。这个context是WatermarkGeneratorSupplier.Context里面能拿到MetricGroup、StreamConfig、ProcessingTimeService之类的运行时资源。这就是为什么你可以在createWatermarkGenerator里拿到 metric 分组、注册自定义指标——很多生产项目就是靠这个把水位线延迟做成了监控面板。WatermarkStrategyMyEvent strategy WatermarkStrategy .MyEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, recordTimestamp) - event.getEventTime()) .withIdleness(Duration.ofSeconds(30));这段代码里的链式调用顺序其实有讲究。withTimestampAssigner返回的是一个新对象withIdleness也是。每个方法都在原来的策略上加一层包装最后交给TimestampsAndWatermarksOperator的是一个组合好的策略。如果你的TimestampAssigner内部有状态比如需要查维表务必保证它是可序列化的因为整个策略对象会被序列化进作业图在 TaskManager 端反序列化执行。我见过有人在匿名内部类里捕获了一个数据库连接池作业提交直接失败。3.3 TimestampsAndWatermarksOperator 的 processElement 走读真正干活的是org.apache.flink.streaming.runtime.operators.TimestampsAndWatermarksOperator。它的processElement是一段特别紧凑的代码但每一步都不能少。Override public void processElement(StreamRecordT element) throws Exception { final long newTimestamp userFunction.extractTimestamp( element.getValue(), element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE); output.collect(element.replace(element.getValue(), newTimestamp)); userFunction.onEvent(element.getValue(), newTimestamp, output); }注意执行顺序先提取时间戳然后立刻把记录带新时间戳往下游发出去最后才调用onEvent。这个顺序非常关键。它意味着当前这条数据本身是在水位线更新之前就发出去的不会出现“数据还没发水位线先跑了导致这条数据被判为迟到”的荒谬情况。还有一个点是element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE。如果上游没有给记录设置时间戳这里会传Long.MIN_VALUE给提取器。你的TimestampAssigner第二个参数拿到Long.MIN_VALUE的时候不要惊讶这只是一个“没有上游时间戳”的标记。如果你在提取器里对第二个参数做算术运算会直接溢出这是个很典型的低级错误。processWatermark就简单得多它把上游来的水位线直接output.emitWatermark(mark)转发下去。但是注意默认情况下TimestampsAndWatermarksOperator会优先使用自己生成的水位线而不是上游的。这就有个容易搞混的场景如果你在一个已经带事件时间的流上再套一层assignTimestampsAndWatermarks那么下游看到的水位线是由新生成的策略决定的上游那条水位线在这里被“截断”了。我见过有人重复分配时间戳导致水位线推进速度完全不符合预期最后发现问题出在中间多做了一次 assign。3.4 StatusWatermarkValve多输入场景下的合成逻辑单输入的水位线就是一条直线往下传没什么好说的。真正复杂的是多输入比如connect、union、多路 join。这时候框架要解决一个问题两个输入各自有各自的水位线下游算子只有一个“当前水位线”的概念怎么合并答案是org.apache.flink.streaming.runtime.io.StatusWatermarkValve。它的核心逻辑是维护一个StreamStatus数组每个输入通道一个槽位记录该通道的最新水位线和是否处于活跃状态。每次某个通道的水位线更新它就重新计算一次取所有活跃通道的水位线最小值如果这个最小值比当前已发出的水位线大就往下游发一条。这里的关键词是“活跃”。如果一个通道长时间没有数据它的水位线会停在原地把整个最小值卡死。StatusWatermarkValve提供的解法是通道状态标记把没数据的通道标记为 IDLE计算最小值的时候跳过它。这就是withIdleness在底层真正的落点。4. 自己写一个 WatermarkGenerator从需求到验证4.1 为什么内置实现有时不够用BoundedOutOfOrdernessWatermarks的问题在于它的乱序容忍度是一个静态常量。你设 5 秒那它就永远是 5 秒。但真实数据流的乱序程度是会变的白天高峰期网络拥塞、乱序严重凌晨低谷时几乎有序。固定值要么在高峰期不够用丢数据要么在低谷期太保守白白增加延迟。一个更聪明的做法是让生成器自己观察并调整。思路不复杂在onEvent里如果新来的数据时间戳小于当前maxTimestamp说明这条数据迟到了迟到幅度就是maxTimestamp - eventTimestamp。把这个幅度记录下来用某种方式聚合成一个“近期观察到的最坏乱序量”然后在onPeriodicEmit里用这个动态值代替静态值。聚合方式我试过两种。一种是简单的滑动窗口最大值维护一个固定长度的时间桶队列取最近 N 个桶里的最大值。另一种是指数衰减每来一条数据就把当前估计值乘一个衰减系数再和新观测值取大。前者的行为更可预测、更容易解释给业务方后者实现更简单、内存占用恒定。生产上我倾向第一种因为“最近一分钟内最坏乱序是 8 秒”这句话能直接跟人解释而衰减系数不太好沟通。4.2 完整实现代码下面这个版本用的是固定时间桶的滑动最大值桶宽和桶数都可以通过构造函数指定实际相当于“最近bucketSize * numBuckets时间内的最大乱序观测值”。import org.apache.flink.api.common.eventtime.Watermark; import org.apache.flink.api.common.eventtime.WatermarkGenerator; import org.apache.flink.api.common.eventtime.WatermarkOutput; public class AdaptiveOutOfOrdernessWatermarksT implements WatermarkGeneratorT { private final long bucketWidthMs; private final long[] buckets; private int currentBucketIndex; private long currentBucketStart; private long maxTimestamp Long.MIN_VALUE; private boolean initialized false; private final long floorLatenessMs; private final long ceilingLatenessMs; public AdaptiveOutOfOrdernessWatermarks(long bucketWidthMs, int numBuckets, long floorLatenessMs, long ceilingLatenessMs) { this.bucketWidthMs bucketWidthMs; this.buckets new long[numBuckets]; this.floorLatenessMs floorLatenessMs; this.ceilingLatenessMs ceilingLatenessMs; } Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { if (!initialized) { initialized true; maxTimestamp eventTimestamp; currentBucketStart eventTimestamp - (eventTimestamp % bucketWidthMs); return; } if (eventTimestamp maxTimestamp) { maxTimestamp eventTimestamp; } else { long lateness maxTimestamp - eventTimestamp; rotateBucketsIfNeeded(eventTimestamp); int idx currentBucketIndex; buckets[idx] Math.max(buckets[idx], lateness); } } private void rotateBucketsIfNeeded(long eventTimestamp) { long bucketStart eventTimestamp - (eventTimestamp % bucketWidthMs); long delta (bucketStart - currentBucketStart) / bucketWidthMs; if (delta 0) { return; } int steps (int) Math.min(delta, buckets.length); for (int i 0; i steps; i) { currentBucketIndex (currentBucketIndex 1) % buckets.length; buckets[currentBucketIndex] 0L; } currentBucketStart bucketStart; } private long currentLatenessBudget() { long max 0L; for (long b : buckets) { if (b max) { max b; } } long budget Math.max(max * 2, floorLatenessMs); return Math.min(budget, ceilingLatenessMs); } Override public void onPeriodicEmit(WatermarkOutput output) { if (!initialized) { return; } output.emitWatermark(new Watermark(maxTimestamp - currentLatenessBudget() - 1)); } }几点说明。max * 2这个系数是留的余量因为观测到的乱序只是“已经发生的”未来可能更糟翻倍是个粗糙但有效的安全边际。floorLatenessMs防止在数据完全有序时预算降到 0 导致水位线过于激进ceilingLatenessMs防止某次异常乱序把水位线永久拖住——注意由于桶会滚动异常值最终会被挤出窗口但设个上限更保险。用的时候这样接DataStreamMyEvent stream env.addSource(source) .assignTimestampsAndWatermarks( WatermarkStrategy .MyEventforGenerator(ctx - new AdaptiveOutOfOrdernessWatermarks(10_000L, 6, 1_000L, 60_000L)) .withTimestampAssigner((e, ts) - e.getEventTime()) );forGenerator接收的是一个WatermarkGeneratorSupplier注意这里传的是 lambda每次算子实例化时会调用一次所以每个 subtask 都有自己独立的生成器实例状态互不共享。这一点很重要生成器的状态是 per-subtask 的不是全局的。如果你的数据分布不均匀不同 subtask 观察到的乱序程度会不一样发出的水位线也会不一样最后在下游合并时取最小值整体效果由最慢的那个决定。4.3 参数怎么定从业务 SLA 反推定参数这件事不要从“我觉得乱序大概几秒”开始而要从业务能接受的延迟倒推。假设业务要求数据在事件发生 30 秒内出现在结果里那么总延迟预算 30 秒里要扣掉网络传输通常 1 到 3 秒、算子处理视逻辑复杂度留 2 到 5 秒、checkpoint 造成的短暂阻塞如果开了 exactly-once留 1 到 2 秒、周期发射间隔最多 200 毫秒可忽略。剩下的才是留给乱序容忍度的空间大概 20 秒左右。然后拿这个预算去比对实际观测到的乱序分布。方法很粗暴但有效先设一个很大的容忍度比如 60 秒跑一天同时把每条数据的迟到幅度打到日志或指标里最后看 P99 是多少。如果 P99 是 6 秒那设 8 到 10 秒就够了不用把 20 秒全用掉。剩下的预算可以作为 buffer 应对突发。提示不要用平均值来定这个参数。乱序分布是典型的长尾分布平均值毫无意义一定要看高分位数。我见过用平均值设成 2 秒结果丢掉 8% 数据的案例。4.4 怎么确认水位线真的按预期在推进代码写完只是第一步验证才是重点。最直接的手段是 Web UI 上算子详情页里的currentInputWatermark和currentOutputWatermark两个指标它们会显示当前 subtask 看到和发出的水位线数值。但这两个数只有瞬时值看不出趋势。更好的做法是自己注册一个 gauge。在自定义生成器的createWatermarkGenerator里能拿到MetricGroup可以这样挂WatermarkStrategy.MyEventforGenerator(ctx - { MetricGroup metrics ctx.getMetricGroup(); AdaptiveOutOfOrdernessWatermarksMyEvent gen new AdaptiveOutOfOrdernessWatermarks(10_000L, 6, 1_000L, 60_000L); metrics.gauge(watermarkLagMs, () - System.currentTimeMillis() - gen.currentWatermark()); return gen; })注意currentWatermark()这个方法在上面那份代码里我没写实际用的时候需要补一个返回maxTimestamp - currentLatenessBudget() - 1的方法出来。有了这个指标你就能在监控面板上直接看到“水位线落后当前系统时间多少毫秒”随时间的变化曲线调参的时候盯着这条曲线看比猜靠谱得多。5. 踩坑实录那些文档里不会写的问题5.1 水位线卡住不动的四类原因这是最高频的问题我按遇到频率排了个序。第一类是某个并行度没数据。假设 source 并行度是 8其中 3 个 subtask 分到的分区暂时没有数据写入它们的生成器就永远停在初始状态发出的水位线是Long.MIN_VALUE。下游算子合并时取最小值结果就是Long.MIN_VALUE所有窗口都不触发。解法是withIdleness把空闲通道标记出来跳过。第二类是反压传导。水位线和数据走同一条通道如果某个下游算子的输入缓冲区满了水位线也发不出去。这种情况在 Web UI 上会看到明显的反压标记但如果你只盯着水位线指标看会觉得莫名其妙地停了。这时候要去查下游算子的处理耗时而不是怀疑生成器。第三类是时间戳单位搞错。TimestampAssigner返回的必须是毫秒。如果你的数据源给的是秒级时间戳10 位数字直接返回就会得到一个 1970 年附近的时间水位线永远在很低的数值徘徊窗口永远不触发。这个问题在测试环境特别隐蔽因为数据量小的时候你可能根本没注意到窗口没出结果。第四类是生成器里的状态没被正确更新。比如你在onEvent里忘了更新maxTimestamp只在某些分支更新那水位线就会一直停在那里。自定义生成器一定要保证onEvent是无条件更新状态的。现象最可能的原因排查手段水位线固定为极小的负数有空闲并行度或分区看各 subtask 的输入记录数开 idleness水位线缓慢但持续落后反压或处理耗时高看反压指标、下游算子耗时水位线数值明显是 1970 年附近时间戳单位不是毫秒打印一条原始记录的时间戳核对水位线周期性回退多路合并逻辑或自定义生成器有 bug检查是否手动 emit 了小值水位线正常但窗口不出结果窗口类型不匹配或触发条件特殊核对窗口 assigner 和 watermark 语义5.2 空闲检测到底该怎么配withIdleness(Duration)的参数是“多久没数据就标记为空闲”。这个值不能设得太小否则正常的处理抖动会被误判为空闲导致水位线在不该推进的时候推进丢掉本该等待的数据。也不能设得太大否则空闲分区的拖累时间会很长。我的经验值是这个值应该大于数据到达的典型间隔的 3 到 5 倍。比如你的分片数据平均每 2 秒来一条那设 10 秒比较合适。如果数据到达本身就很稀疏比如每 5 分钟一条那这个机制其实不太适用你可能需要考虑换一种分片策略或者干脆不用事件时间窗口。还有一个坑withIdleness只对输入做空闲标记它标记的是“这个输入通道空闲”。但如果你有多个 source其中一个 source 整体停了它的空闲标记要能传播到下游才行。这条路径在不同版本里行为有过调整建议实际测试一下你的版本在“一路 source 完全停止”时的表现别假设它一定能正常工作。5.3 迟到数据的三种处置方式对比水位线越过之后到达的数据默认是被丢弃的。这在很多场景下不可接受Flink 提供了三条路。第一种是加大乱序容忍度。这是最省事的代价是整体延迟变大。适合延迟不敏感、准确性要求高的场景。第二种是开启允许迟到。在窗口上调用allowedLateness(Duration)窗口触发之后不会立刻销毁而是保留一段时间期间每来一条迟到数据就重新触发一次计算。代价是每个窗口要额外维护一份状态窗口越多内存压力越大。要注意这会带来“同一个窗口多次输出”的行为下游必须能处理重复结果。第三种是侧输出迟到数据。用sideOutputLateData(OutputTag)把迟到数据分流出去单独做补偿处理。这是最灵活的方式适合需要精确对账的场景。代价是你得为这条侧流写额外的处理逻辑。实际项目里我经常混用窗口上开一个较小的allowedLateness比如 30 秒覆盖绝大多数情况同时挂一个侧输出兜住极端迟到数据侧流写入一个单独的表离线任务再和主结果做一次对账合并。这样既不牺牲太多实时性也不会真的丢数据。5.4 自定义生成器的两个隐蔽陷阱第一个陷阱是每 subtask 独立状态带来的不一致。前面提过生成器实例是 per-subtask 的。如果你的 source 数据分布有倾斜某个 subtask 拿到的时间戳范围比另一个宽它们发出的水位线差异可能很大。下游合并取最小值整体进度由最慢的那个决定。这在调参时会造成困扰你在某个 subtask 上看到水位线推进很快但整体没动会误以为是别的问题。排查时一定要同时看所有 subtask 的水位线而不是只看一个。第二个陷阱是在 onEvent 里直接 emit 水位线。技术上完全可行但会引入一个微妙的问题这条水位线是在处理当前数据的中间发出的而当前数据的处理还没结束虽然数据本身已经 collect 出去了。如果下游有多个输入这条早发的水位线可能先于其他输入的数据到达造成短暂的“水位线跳变”。打点型生成器必须这么做但如果做周期型一定要忍住别在onEvent里 emit。6. 几个容易被忽略但影响很大的工程细节6.1 时间戳精度与边界的那点事前面讲- 1的时候提了一句这里展开说。窗口的触发条件在源码里是timerService.registerEventTimeTimer(window.maxTimestamp())而Window.maxTimestamp()的默认实现是endTime - 1。对于[10:00:00, 10:01:00)这个窗口maxTimestamp()是10:00:59.999也就是10:00:59加上 999 毫秒——在毫秒精度的long表示里就是endTime - 1。所以要让这个窗口触发水位线必须 10:00:59.999。而BoundedOutOfOrdernessWatermarks发出的水位线是maxTimestamp - outOfOrderness - 1。假设乱序容忍度 0一条时间戳正好是10:00:59.999的数据到达发出的水位线是10:00:59.998窗口不触发。要等下一条时间戳 10:01:00.000的数据到达水位线变成10:00:59.999窗口才触发。这就是为什么“以最大时间戳结尾的那批数据所在的窗口”需要额外一条后续数据才能关闭这个行为经常让人困惑。如果你的时间戳精度是秒比如long存的是秒那么所有时间戳都是整秒maxTimestamp()是endTime - 1你会发现窗口永远差一点。这种情况下必须把时间戳乘以 1000 转成毫秒别无他法。6.2 并行度、分区与水位线传播的关系水位线在算子之间是广播的也就是说上游每个 subtask 发出的水位线会被发送到下游所有 subtask。下游每个 subtask 收到 N 条水位线N 是上游并行度合并取最小值。这个机制带来两个后果。一是水位线的传输量随并行度线性增长虽然每条只有几十字节但高频发射还是有开销这也是为什么周期型的默认间隔是 200 毫秒而不是 10 毫秒。二是木桶效应特别明显上游并行度越大越有可能出现某一两个慢的分区把整体水位线拖住。所以调大并行度的时候要留意水位线的表现。我在一个项目里把 source 并行度从 4 调到 16吞吐上去了但端到端延迟涨了将近一倍原因就是 16 个分区里总有那么一两个分区数据稀疏withIdleness配得又不够激进水位线整体推进被拖慢。后来把 idleness 从 30 秒调到 8 秒才恢复正常。6.3 checkpoint 恢复之后水位线在哪里这是个容易让人心里没底的问题。WatermarkGenerator的内部状态比如maxTimestamp默认是不参与 checkpoint 的。因为它不是CheckpointedFunction也没有实现任何快照接口。这意味着作业从 checkpoint 恢复之后生成器的状态是全新的maxTimestamp会被重置到初始值。那会发生什么恢复后的第一批数据会把maxTimestamp重新推上去然后onPeriodicEmit发出一批水位线。由于恢复后的水位线从很小的值开始爬升下游的窗口定时器这些定时器是注册在 keyed state 里的会随 checkpoint 恢复在恢复后要等到水位线重新爬过它们触发点才会执行。如果你的 checkpoint 间隔比较长或者恢复后数据有断档可能出现“恢复后一段时间内所有窗口都不触发然后突然集中触发一大批”的现象。这个行为本身是语义正确的at-least-once 或者 exactly-once 都允许重算但如果你的下游对突发流量敏感需要提前做容量准备。注意如果你确实需要生成器状态参与恢复比如那个自适应乱序估计器恢复后重新学习会有个冷启动期可以自己实现CheckpointedFunction在initializeState里恢复maxTimestamp和桶数组。不过要小心恢复的maxTimestamp如果比当前实际数据的时间戳大很多会导致恢复后立刻把水位线推得很高把新数据全判成迟到。这个权衡得自己拿捏。6.4 关于 final watermark 的一点观察算子关闭的时候默认会补发一条Watermark.MAX_WATERMARK目的是让下游所有还挂着的窗口立刻结算并输出避免作业停止时丢掉最后一批结果。这个行为在不同版本之间有过调整有些版本可以通过参数关掉。我的建议是自己动手验证一下你用的版本到底是什么行为方法是写一个最简单的本地作业一个 source 发几条带时间戳的数据一个滑动窗口做聚合。跑完之后看输出如果最后那批本该被窗口持有的数据被输出出来了说明 final watermark 起了作用。这个验证十分钟就能做完但能帮你避免很多“作业停了结果少了一截”的困惑。另外提醒一句如果你在管道里串了多个assignTimestampsAndWatermarks虽然不推荐但有时为了在不同阶段用不同策略会这么干最终水位线是被最后一个生成器决定的中间那些策略的 final watermark 行为可能互相影响测试的时候要把整条链都跑通。我自己在这个方向上折腾了挺久最大的体会是Watermark 的难点从来不在概念概念三句话就能讲完难点在于它是一条贯穿数据源、算子链、网络传输、状态恢复的完整链路任何一个环节出问题表现都是“窗口不触发”这一个症状。所以排查的时候不要只盯着生成器看要从数据源的时间戳一路往下捋看每个 subtask 的输入水位线、输出水位线、下游的合并结果一级一级对通常十分钟内就能定位。至于参数怎么调别信任何“最佳实践”里给的固定数值那些数字都是在特定数据分布下测出来的你的数据分布不一样答案就不一样老老实实打点、观察、调、再观察这才是唯一靠谱的路子。 
长河君 · 2026/10/28 20:09:52 · 阅读 1.2k Flink Watermark源码解析：WatermarkGenerator与窗口触发机制
做流处理这几年被问得最多的一类问题几乎都和 Flink 的 Watermark 有关。不是“Watermark 是什么”这种概念题而是“我明明设置了 WatermarkGenerator窗口为什么就是不触发”“水位线卡在某个值上一动不动”“并行度调大之后延迟突然涨了好几倍”。这些问题的共同点是只看官方文档里的那几行示例代码看不出答案必须翻到 WatermarkGenerator 的实现里看清楚它被谁调用、什么时候调用、调用之后那根水位线又经过了几层包装才流到下游。这篇内容就是把我自己读 Flink 源码的路径重新走一遍从 WatermarkGenerator 这个只有两个方法的接口切入一路跟到 TimestampsAndWatermarksOperator再到多输入场景下的水位线裁决逻辑中间穿插我自己写自定义生成器时踩过的坑和验证手段。适合已经能跑通 Flink 作业、但对事件时间和水位线心里没底的兄弟也适合想动手改 WatermarkGenerator 但不确定副作用的人。1. 先把问题摆到桌面上为什么非要有这根水位线1.1 从窗口触发的那个死结说起假设你在做一个按分钟统计的实时看板数据来自一堆客户端上报每条记录自带一个发生时间戳。业务上你要的是“按照数据真正发生的时间每 1 分钟聚一次”也就是事件时间语义。现在问题来了你要等到什么时候才敢把 10:00 到 10:01 这一分钟的窗口算出来并输出结果这里有个绕不过去的死结。数据在网络里是会乱序的10:00:59 的那条记录可能比 10:00:03 的那条晚到十几秒。如果你一看到时间戳越过 10:01:00 就把窗口结算了那后面姗姗来迟的记录就只能被丢掉或者重算。反过来如果你一直等等到“确定不会再有 10:01 之前的数据了”那这个“确定”本身就需要一个外部信号——流处理系统自己没法凭空知道未来还有没有数据。Watermark 就是这个外部信号的形式化表达。它是一条特殊的数据记录只带一个时间戳语义是“事件时间小于等于这个时间戳的数据我认为已经全部到齐了。”注意这里用的是“我认为”不是“我保证”这个措辞差别很关键后面讲迟到数据的时候还会回到这一点。所以整个机制的逻辑链条是数据自带时间戳 → 用一个生成策略持续地抬升水位线 → 算子收到水位线后把“已到期”的定时器比如窗口结束时间触发掉 → 结果输出。水位线的推进速度直接决定了你的结果延迟和结果准确性之间的平衡点而这个平衡点就是 WatermarkGenerator 要负责的事情。1.2 Watermark 在源码里到底长什么样很多人第一次翻到org.apache.flink.streaming.api.watermark.Watermark这个类的时候都会有点意外因为它实在是太简单了。它继承自StreamElement和StreamRecord、WatermarkStatus是平级的核心就只有一个long timestamp字段外加一个常量MAX_WATERMARK new Watermark(Long.MAX_VALUE)。这个设计其实挺说明问题的。水位线在运行时就是一条普通元素跟着数据记录一起在算子之间流动走的是同一套网络传输和反压机制。它不是某种控制平面的元数据也不是在某个中心节点维护的全局变量。这意味着两件事第一水位线会像数据一样被缓冲、被反压卡住如果一个算子处理慢它上游的水位线也会跟着堵第二水位线在跨算子传递时是有分区概念的这一点在后面讲并行度的时候会展开。另外一个容易被忽略的细节是Watermark类没有实现任何比较逻辑也没有单调性校验。也就是说框架层面并不会阻止你发出一个比上一条更小的水位线虽然那样做在语义上是错误并且会导致下游定时器状态混乱。这个约束是靠WatermarkOutput那一层的包装来兜底的而不是靠Watermark自己。理解这一点很重要你写自定义生成器时如果直接拿到WatermarkOutput就乱发很可能不会立刻报错而是过一段时间之后在某个窗口算子里看到莫名其妙的结果。1.3 三条语义保证和它们各自的代价把水位线的语义拆开实际上有三条相互关联的保证在起作用。第一条是单调性同一个输出通道上的水位线只能增不能减。这条保证让下游可以安全地用一个long变量记住“当前水位线”不用做任何比较。代价是如果一个分区长时间没有数据它的水位线就不动会把整个下游卡住——这直接催生了空闲检测idleness机制。第二条是全序到偏序的转换单个分区内的水位线是单调的但多个上游分区之间没有顺序关系。下游算子必须把它们合并成一个合并规则是取最小值。取最小值是为了保证“所有上游都说这个时间点之前的数据到齐了”才成立任何一路没到齐就不能推进。代价是只要有一路慢全局都慢这是水位线最容易出问题的地方。第三条是不保证不丢数据水位线只表达“我认为到齐了”不表达“我保证到齐了”。越过水位线之后到达的数据就是迟到数据默认会被丢弃。代价是你要么加大乱序容忍度换取准确性、牺牲延迟要么开启allowedLateness让窗口多留一会儿要么用侧输出把迟到数据接出来单独处理。这三条放在一起看Watermark 的整个设计取向就清楚了它不追求绝对正确它追求的是在有限延迟内给出一个可以量化的近似结果把“要多准”这个决策权交给使用者。而这个决策权的具体承载物就是接下来要拆的WatermarkGenerator。2. WatermarkGenerator 接口两个方法背后的调用时序2.1 onEvent 与 onPeriodicEmit 分别被谁调用打开org.apache.flink.api.common.eventtime.WatermarkGenerator你会看到一个特别克制的接口定义只有两个方法onEvent(T event, long eventTimestamp, WatermarkOutput output)和onPeriodicEmit(WatermarkOutput output)。没有 init没有 close没有获取当前水位线的 getter。这种极简设计是有意为之的接口越窄框架对实现的假设就越少。onEvent的调用时机很明确每处理一条数据就调一次在TimestampsAndWatermarksOperator.processElement里紧跟在时间戳提取之后。注意传入的eventTimestamp是已经由TimestampAssigner提取好的值不是你原始数据里的字段所以如果你在onEvent里再去解析一遍字段纯属浪费。onPeriodicEmit的调用时机就有讲究了。它是由算子内部基于处理时间注册的一个定时器周期性触发的间隔由pipeline.auto-watermark-interval控制默认 200 毫秒。这里的关键词是“处理时间”——也就是说即使你的数据流完全静止、一条数据都不来只要这个定时器还在跑onPeriodicEmit就会被调用。这解释了一个常见困惑“我的流里没数据了为什么下游还在收到水位线”答案是周期型生成器在空转。还有一个隐蔽的细节当auto-watermark-interval被设置为 0 的时候周期定时器根本不会被注册onPeriodicEmit永远不会被调用。这就是为什么纯打点型punctuated生成器必须把这个间隔设为 0否则就是白白浪费一次回调。2.2 周期型和打点型本质是同一接口的两种用法官方文档把水位线生成分成“周期型”和“打点型”两类但从源码角度看这两者并没有对应的两个接口它们只是同一个WatermarkGenerator接口的两种填充方式。周期型的做法是onEvent里只更新内部状态比如记录见过的最大时间戳真正的emitWatermark放在onPeriodicEmit里。好处是发水位线的频率和数据量解耦吞吐高的时候不会因为每条数据都发一次水位线而把网络打满。坏处是水位线的推进有最多一个 interval 的延迟精细控制做不到。打点型的做法反过来onEvent里判断某个条件比如遇到一个特殊标记事件、或者跨过了某个整数边界条件满足就立刻output.emitWatermark(...)onPeriodicEmit留空。好处是响应及时语义上可以和业务事件绑死。坏处是如果条件设计得不好可能出现长时间不发水位线导致下游窗口不触发或者发得太频繁造成额外开销。维度周期型打点型状态更新位置onEventonEvent水位线发射位置onPeriodicEmitonEvent发射频率决定因素处理时间间隔数据内容数据静止时是否推进是会空转否典型配置interval 默认 200msinterval 设为 0适合场景绝大多数无特殊标记的流有明确结束标记或分片边界的流实际项目里我基本都用周期型因为绝大多数数据源没有可靠的“分片结束”标记硬凑一个打点条件反而容易出错。只有在处理那种带明确批次结束标记的数据比如文件导入、MQ 里的批次完成消息时打点型才有明显优势。2.3 WatermarkOutput被包装了一层的那道闸门WatermarkOutput这个接口名字看着像输出端实际用起来更像一道闸门。它只有一个emitWatermark(Watermark watermark)方法和一个markIdle()新版是emitWatermarkStatus(WatermarkStatus)。为什么需要这层抽象因为生成器发出去的水位线不一定会原封不动地流到网络上去。在算子内部这个output可能是一个WatermarkOutputMultiplexer.Output也可能是一个带着空闲状态跟踪的包装。最典型的是WatermarkOutputMultiplexer的场景当一个算子的输出被链式chained连接到多个下游输入时生成器只发一次水位线多路复用器要把它记录下来然后对每一路下游分别计算“应该看到的水位线”再分别转发。这里有个很容易忽略的点WatermarkOutputMultiplexer在合并多路输出时取的也是最小值。也就是说如果你在一个ProcessFunction里做了分流两路下游的处理速度不一样慢的那一路会拖住快的那一路看到的水位线吗答案是会因为多路复用器是取所有 active 输出的最小值来算出“组合水位线”的。这是从正确性出发的保守设计实际表现就是分流之后整体水位线推进变慢。我在一个双流 join 的项目里被这个坑过一次排查了半天才意识到是其中一路下游反压导致水位线整体停滞。3. 沿源码往下走从一个接口到真正的水位线3.1 BoundedOutOfOrdernessWatermarks 到底做了什么几乎所有人第一个接触的实现都是BoundedOutOfOrdernessWatermarks也就是WatermarkStrategy.forBoundedOutOfOrderness(Duration)背后那个类。它的代码短得可怜但每一行都有讲究。public class BoundedOutOfOrdernessWatermarksT implements WatermarkGeneratorT { /** The maximum timestamp encountered so far in the stream. */ private long maxTimestamp; /** The maximum out-of-orderness that this watermark generator assumes. */ private final long outOfOrdernessMillis; public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) { checkNotNull(maxOutOfOrderness, maxOutOfOrderness); checkArgument(!maxOutOfOrderness.isNegative(), maxOutOfOrderness cannot be negative); this.outOfOrdernessMillis maxOutOfOrderness.toMillis(); // start so that our lowest watermark would be Long.MIN_VALUE. this.maxTimestamp Long.MIN_VALUE outOfOrdernessMillis 1; } Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { maxTimestamp Math.max(maxTimestamp, eventTimestamp); } Override public void onPeriodicEmit(WatermarkOutput output) { output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1)); } }先看构造函数里那个初始值Long.MIN_VALUE outOfOrdernessMillis 1。第一眼看上去很奇怪为什么要加个outOfOrdernessMillis 1原因是这一行同时防了两件事。一是防止下溢如果maxTimestamp从Long.MIN_VALUE开始onPeriodicEmit里算Long.MIN_VALUE - outOfOrdernessMillis - 1就会直接溢出成一个巨大的正数发出去一条Long.MAX_VALUE级别的水位线下游所有窗口瞬间全部触发。二是保证初始水位线恰好落在Long.MIN_VALUE上这样“还没收到任何数据”的状态不会误触发任何定时器。再看onPeriodicEmit里那个- 1。这个才是最容易看漏的地方。假设乱序容忍度是 0AscendingTimestampsWatermarks的情况那么发出的水位线就是maxTimestamp - 1而不是maxTimestamp。为什么要减这个 1因为 Watermark 的语义是“小于等于该时间戳的数据已经到齐”而窗口的触发条件是watermark window.maxTimestamp()其中window.maxTimestamp()对于时间窗口来说是endTime - 1。如果不减 1那么当水位线正好等于maxTimestamp时maxTimestamp本身那条数据对应的窗口边界可能还没到就会出现“水位线到了但窗口没触发”或者“窗口提前触发”的偏差。这个 1 是边界对齐的修正项不是拍脑袋。3.2 时间戳提取器和生成器的组装过程单独一个WatermarkGenerator是不能工作的它必须和一个TimestampAssigner一起打包成WatermarkStrategy。而这个打包动作发生在作业图构建阶段不是在运行时。WatermarkStrategy的实现类WatermarkStrategy.WithTimestampsWatermarks里有个createWatermarkGenerator方法它会调用WatermarkGeneratorSupplier.createWatermarkGenerator(context)。这个context是WatermarkGeneratorSupplier.Context里面能拿到MetricGroup、StreamConfig、ProcessingTimeService之类的运行时资源。这就是为什么你可以在createWatermarkGenerator里拿到 metric 分组、注册自定义指标——很多生产项目就是靠这个把水位线延迟做成了监控面板。WatermarkStrategyMyEvent strategy WatermarkStrategy .MyEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, recordTimestamp) - event.getEventTime()) .withIdleness(Duration.ofSeconds(30));这段代码里的链式调用顺序其实有讲究。withTimestampAssigner返回的是一个新对象withIdleness也是。每个方法都在原来的策略上加一层包装最后交给TimestampsAndWatermarksOperator的是一个组合好的策略。如果你的TimestampAssigner内部有状态比如需要查维表务必保证它是可序列化的因为整个策略对象会被序列化进作业图在 TaskManager 端反序列化执行。我见过有人在匿名内部类里捕获了一个数据库连接池作业提交直接失败。3.3 TimestampsAndWatermarksOperator 的 processElement 走读真正干活的是org.apache.flink.streaming.runtime.operators.TimestampsAndWatermarksOperator。它的processElement是一段特别紧凑的代码但每一步都不能少。Override public void processElement(StreamRecordT element) throws Exception { final long newTimestamp userFunction.extractTimestamp( element.getValue(), element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE); output.collect(element.replace(element.getValue(), newTimestamp)); userFunction.onEvent(element.getValue(), newTimestamp, output); }注意执行顺序先提取时间戳然后立刻把记录带新时间戳往下游发出去最后才调用onEvent。这个顺序非常关键。它意味着当前这条数据本身是在水位线更新之前就发出去的不会出现“数据还没发水位线先跑了导致这条数据被判为迟到”的荒谬情况。还有一个点是element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE。如果上游没有给记录设置时间戳这里会传Long.MIN_VALUE给提取器。你的TimestampAssigner第二个参数拿到Long.MIN_VALUE的时候不要惊讶这只是一个“没有上游时间戳”的标记。如果你在提取器里对第二个参数做算术运算会直接溢出这是个很典型的低级错误。processWatermark就简单得多它把上游来的水位线直接output.emitWatermark(mark)转发下去。但是注意默认情况下TimestampsAndWatermarksOperator会优先使用自己生成的水位线而不是上游的。这就有个容易搞混的场景如果你在一个已经带事件时间的流上再套一层assignTimestampsAndWatermarks那么下游看到的水位线是由新生成的策略决定的上游那条水位线在这里被“截断”了。我见过有人重复分配时间戳导致水位线推进速度完全不符合预期最后发现问题出在中间多做了一次 assign。3.4 StatusWatermarkValve多输入场景下的合成逻辑单输入的水位线就是一条直线往下传没什么好说的。真正复杂的是多输入比如connect、union、多路 join。这时候框架要解决一个问题两个输入各自有各自的水位线下游算子只有一个“当前水位线”的概念怎么合并答案是org.apache.flink.streaming.runtime.io.StatusWatermarkValve。它的核心逻辑是维护一个StreamStatus数组每个输入通道一个槽位记录该通道的最新水位线和是否处于活跃状态。每次某个通道的水位线更新它就重新计算一次取所有活跃通道的水位线最小值如果这个最小值比当前已发出的水位线大就往下游发一条。这里的关键词是“活跃”。如果一个通道长时间没有数据它的水位线会停在原地把整个最小值卡死。StatusWatermarkValve提供的解法是通道状态标记把没数据的通道标记为 IDLE计算最小值的时候跳过它。这就是withIdleness在底层真正的落点。4. 自己写一个 WatermarkGenerator从需求到验证4.1 为什么内置实现有时不够用BoundedOutOfOrdernessWatermarks的问题在于它的乱序容忍度是一个静态常量。你设 5 秒那它就永远是 5 秒。但真实数据流的乱序程度是会变的白天高峰期网络拥塞、乱序严重凌晨低谷时几乎有序。固定值要么在高峰期不够用丢数据要么在低谷期太保守白白增加延迟。一个更聪明的做法是让生成器自己观察并调整。思路不复杂在onEvent里如果新来的数据时间戳小于当前maxTimestamp说明这条数据迟到了迟到幅度就是maxTimestamp - eventTimestamp。把这个幅度记录下来用某种方式聚合成一个“近期观察到的最坏乱序量”然后在onPeriodicEmit里用这个动态值代替静态值。聚合方式我试过两种。一种是简单的滑动窗口最大值维护一个固定长度的时间桶队列取最近 N 个桶里的最大值。另一种是指数衰减每来一条数据就把当前估计值乘一个衰减系数再和新观测值取大。前者的行为更可预测、更容易解释给业务方后者实现更简单、内存占用恒定。生产上我倾向第一种因为“最近一分钟内最坏乱序是 8 秒”这句话能直接跟人解释而衰减系数不太好沟通。4.2 完整实现代码下面这个版本用的是固定时间桶的滑动最大值桶宽和桶数都可以通过构造函数指定实际相当于“最近bucketSize * numBuckets时间内的最大乱序观测值”。import org.apache.flink.api.common.eventtime.Watermark; import org.apache.flink.api.common.eventtime.WatermarkGenerator; import org.apache.flink.api.common.eventtime.WatermarkOutput; public class AdaptiveOutOfOrdernessWatermarksT implements WatermarkGeneratorT { private final long bucketWidthMs; private final long[] buckets; private int currentBucketIndex; private long currentBucketStart; private long maxTimestamp Long.MIN_VALUE; private boolean initialized false; private final long floorLatenessMs; private final long ceilingLatenessMs; public AdaptiveOutOfOrdernessWatermarks(long bucketWidthMs, int numBuckets, long floorLatenessMs, long ceilingLatenessMs) { this.bucketWidthMs bucketWidthMs; this.buckets new long[numBuckets]; this.floorLatenessMs floorLatenessMs; this.ceilingLatenessMs ceilingLatenessMs; } Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { if (!initialized) { initialized true; maxTimestamp eventTimestamp; currentBucketStart eventTimestamp - (eventTimestamp % bucketWidthMs); return; } if (eventTimestamp maxTimestamp) { maxTimestamp eventTimestamp; } else { long lateness maxTimestamp - eventTimestamp; rotateBucketsIfNeeded(eventTimestamp); int idx currentBucketIndex; buckets[idx] Math.max(buckets[idx], lateness); } } private void rotateBucketsIfNeeded(long eventTimestamp) { long bucketStart eventTimestamp - (eventTimestamp % bucketWidthMs); long delta (bucketStart - currentBucketStart) / bucketWidthMs; if (delta 0) { return; } int steps (int) Math.min(delta, buckets.length); for (int i 0; i steps; i) { currentBucketIndex (currentBucketIndex 1) % buckets.length; buckets[currentBucketIndex] 0L; } currentBucketStart bucketStart; } private long currentLatenessBudget() { long max 0L; for (long b : buckets) { if (b max) { max b; } } long budget Math.max(max * 2, floorLatenessMs); return Math.min(budget, ceilingLatenessMs); } Override public void onPeriodicEmit(WatermarkOutput output) { if (!initialized) { return; } output.emitWatermark(new Watermark(maxTimestamp - currentLatenessBudget() - 1)); } }几点说明。max * 2这个系数是留的余量因为观测到的乱序只是“已经发生的”未来可能更糟翻倍是个粗糙但有效的安全边际。floorLatenessMs防止在数据完全有序时预算降到 0 导致水位线过于激进ceilingLatenessMs防止某次异常乱序把水位线永久拖住——注意由于桶会滚动异常值最终会被挤出窗口但设个上限更保险。用的时候这样接DataStreamMyEvent stream env.addSource(source) .assignTimestampsAndWatermarks( WatermarkStrategy .MyEventforGenerator(ctx - new AdaptiveOutOfOrdernessWatermarks(10_000L, 6, 1_000L, 60_000L)) .withTimestampAssigner((e, ts) - e.getEventTime()) );forGenerator接收的是一个WatermarkGeneratorSupplier注意这里传的是 lambda每次算子实例化时会调用一次所以每个 subtask 都有自己独立的生成器实例状态互不共享。这一点很重要生成器的状态是 per-subtask 的不是全局的。如果你的数据分布不均匀不同 subtask 观察到的乱序程度会不一样发出的水位线也会不一样最后在下游合并时取最小值整体效果由最慢的那个决定。4.3 参数怎么定从业务 SLA 反推定参数这件事不要从“我觉得乱序大概几秒”开始而要从业务能接受的延迟倒推。假设业务要求数据在事件发生 30 秒内出现在结果里那么总延迟预算 30 秒里要扣掉网络传输通常 1 到 3 秒、算子处理视逻辑复杂度留 2 到 5 秒、checkpoint 造成的短暂阻塞如果开了 exactly-once留 1 到 2 秒、周期发射间隔最多 200 毫秒可忽略。剩下的才是留给乱序容忍度的空间大概 20 秒左右。然后拿这个预算去比对实际观测到的乱序分布。方法很粗暴但有效先设一个很大的容忍度比如 60 秒跑一天同时把每条数据的迟到幅度打到日志或指标里最后看 P99 是多少。如果 P99 是 6 秒那设 8 到 10 秒就够了不用把 20 秒全用掉。剩下的预算可以作为 buffer 应对突发。提示不要用平均值来定这个参数。乱序分布是典型的长尾分布平均值毫无意义一定要看高分位数。我见过用平均值设成 2 秒结果丢掉 8% 数据的案例。4.4 怎么确认水位线真的按预期在推进代码写完只是第一步验证才是重点。最直接的手段是 Web UI 上算子详情页里的currentInputWatermark和currentOutputWatermark两个指标它们会显示当前 subtask 看到和发出的水位线数值。但这两个数只有瞬时值看不出趋势。更好的做法是自己注册一个 gauge。在自定义生成器的createWatermarkGenerator里能拿到MetricGroup可以这样挂WatermarkStrategy.MyEventforGenerator(ctx - { MetricGroup metrics ctx.getMetricGroup(); AdaptiveOutOfOrdernessWatermarksMyEvent gen new AdaptiveOutOfOrdernessWatermarks(10_000L, 6, 1_000L, 60_000L); metrics.gauge(watermarkLagMs, () - System.currentTimeMillis() - gen.currentWatermark()); return gen; })注意currentWatermark()这个方法在上面那份代码里我没写实际用的时候需要补一个返回maxTimestamp - currentLatenessBudget() - 1的方法出来。有了这个指标你就能在监控面板上直接看到“水位线落后当前系统时间多少毫秒”随时间的变化曲线调参的时候盯着这条曲线看比猜靠谱得多。5. 踩坑实录那些文档里不会写的问题5.1 水位线卡住不动的四类原因这是最高频的问题我按遇到频率排了个序。第一类是某个并行度没数据。假设 source 并行度是 8其中 3 个 subtask 分到的分区暂时没有数据写入它们的生成器就永远停在初始状态发出的水位线是Long.MIN_VALUE。下游算子合并时取最小值结果就是Long.MIN_VALUE所有窗口都不触发。解法是withIdleness把空闲通道标记出来跳过。第二类是反压传导。水位线和数据走同一条通道如果某个下游算子的输入缓冲区满了水位线也发不出去。这种情况在 Web UI 上会看到明显的反压标记但如果你只盯着水位线指标看会觉得莫名其妙地停了。这时候要去查下游算子的处理耗时而不是怀疑生成器。第三类是时间戳单位搞错。TimestampAssigner返回的必须是毫秒。如果你的数据源给的是秒级时间戳10 位数字直接返回就会得到一个 1970 年附近的时间水位线永远在很低的数值徘徊窗口永远不触发。这个问题在测试环境特别隐蔽因为数据量小的时候你可能根本没注意到窗口没出结果。第四类是生成器里的状态没被正确更新。比如你在onEvent里忘了更新maxTimestamp只在某些分支更新那水位线就会一直停在那里。自定义生成器一定要保证onEvent是无条件更新状态的。现象最可能的原因排查手段水位线固定为极小的负数有空闲并行度或分区看各 subtask 的输入记录数开 idleness水位线缓慢但持续落后反压或处理耗时高看反压指标、下游算子耗时水位线数值明显是 1970 年附近时间戳单位不是毫秒打印一条原始记录的时间戳核对水位线周期性回退多路合并逻辑或自定义生成器有 bug检查是否手动 emit 了小值水位线正常但窗口不出结果窗口类型不匹配或触发条件特殊核对窗口 assigner 和 watermark 语义5.2 空闲检测到底该怎么配withIdleness(Duration)的参数是“多久没数据就标记为空闲”。这个值不能设得太小否则正常的处理抖动会被误判为空闲导致水位线在不该推进的时候推进丢掉本该等待的数据。也不能设得太大否则空闲分区的拖累时间会很长。我的经验值是这个值应该大于数据到达的典型间隔的 3 到 5 倍。比如你的分片数据平均每 2 秒来一条那设 10 秒比较合适。如果数据到达本身就很稀疏比如每 5 分钟一条那这个机制其实不太适用你可能需要考虑换一种分片策略或者干脆不用事件时间窗口。还有一个坑withIdleness只对输入做空闲标记它标记的是“这个输入通道空闲”。但如果你有多个 source其中一个 source 整体停了它的空闲标记要能传播到下游才行。这条路径在不同版本里行为有过调整建议实际测试一下你的版本在“一路 source 完全停止”时的表现别假设它一定能正常工作。5.3 迟到数据的三种处置方式对比水位线越过之后到达的数据默认是被丢弃的。这在很多场景下不可接受Flink 提供了三条路。第一种是加大乱序容忍度。这是最省事的代价是整体延迟变大。适合延迟不敏感、准确性要求高的场景。第二种是开启允许迟到。在窗口上调用allowedLateness(Duration)窗口触发之后不会立刻销毁而是保留一段时间期间每来一条迟到数据就重新触发一次计算。代价是每个窗口要额外维护一份状态窗口越多内存压力越大。要注意这会带来“同一个窗口多次输出”的行为下游必须能处理重复结果。第三种是侧输出迟到数据。用sideOutputLateData(OutputTag)把迟到数据分流出去单独做补偿处理。这是最灵活的方式适合需要精确对账的场景。代价是你得为这条侧流写额外的处理逻辑。实际项目里我经常混用窗口上开一个较小的allowedLateness比如 30 秒覆盖绝大多数情况同时挂一个侧输出兜住极端迟到数据侧流写入一个单独的表离线任务再和主结果做一次对账合并。这样既不牺牲太多实时性也不会真的丢数据。5.4 自定义生成器的两个隐蔽陷阱第一个陷阱是每 subtask 独立状态带来的不一致。前面提过生成器实例是 per-subtask 的。如果你的 source 数据分布有倾斜某个 subtask 拿到的时间戳范围比另一个宽它们发出的水位线差异可能很大。下游合并取最小值整体进度由最慢的那个决定。这在调参时会造成困扰你在某个 subtask 上看到水位线推进很快但整体没动会误以为是别的问题。排查时一定要同时看所有 subtask 的水位线而不是只看一个。第二个陷阱是在 onEvent 里直接 emit 水位线。技术上完全可行但会引入一个微妙的问题这条水位线是在处理当前数据的中间发出的而当前数据的处理还没结束虽然数据本身已经 collect 出去了。如果下游有多个输入这条早发的水位线可能先于其他输入的数据到达造成短暂的“水位线跳变”。打点型生成器必须这么做但如果做周期型一定要忍住别在onEvent里 emit。6. 几个容易被忽略但影响很大的工程细节6.1 时间戳精度与边界的那点事前面讲- 1的时候提了一句这里展开说。窗口的触发条件在源码里是timerService.registerEventTimeTimer(window.maxTimestamp())而Window.maxTimestamp()的默认实现是endTime - 1。对于[10:00:00, 10:01:00)这个窗口maxTimestamp()是10:00:59.999也就是10:00:59加上 999 毫秒——在毫秒精度的long表示里就是endTime - 1。所以要让这个窗口触发水位线必须 10:00:59.999。而BoundedOutOfOrdernessWatermarks发出的水位线是maxTimestamp - outOfOrderness - 1。假设乱序容忍度 0一条时间戳正好是10:00:59.999的数据到达发出的水位线是10:00:59.998窗口不触发。要等下一条时间戳 10:01:00.000的数据到达水位线变成10:00:59.999窗口才触发。这就是为什么“以最大时间戳结尾的那批数据所在的窗口”需要额外一条后续数据才能关闭这个行为经常让人困惑。如果你的时间戳精度是秒比如long存的是秒那么所有时间戳都是整秒maxTimestamp()是endTime - 1你会发现窗口永远差一点。这种情况下必须把时间戳乘以 1000 转成毫秒别无他法。6.2 并行度、分区与水位线传播的关系水位线在算子之间是广播的也就是说上游每个 subtask 发出的水位线会被发送到下游所有 subtask。下游每个 subtask 收到 N 条水位线N 是上游并行度合并取最小值。这个机制带来两个后果。一是水位线的传输量随并行度线性增长虽然每条只有几十字节但高频发射还是有开销这也是为什么周期型的默认间隔是 200 毫秒而不是 10 毫秒。二是木桶效应特别明显上游并行度越大越有可能出现某一两个慢的分区把整体水位线拖住。所以调大并行度的时候要留意水位线的表现。我在一个项目里把 source 并行度从 4 调到 16吞吐上去了但端到端延迟涨了将近一倍原因就是 16 个分区里总有那么一两个分区数据稀疏withIdleness配得又不够激进水位线整体推进被拖慢。后来把 idleness 从 30 秒调到 8 秒才恢复正常。6.3 checkpoint 恢复之后水位线在哪里这是个容易让人心里没底的问题。WatermarkGenerator的内部状态比如maxTimestamp默认是不参与 checkpoint 的。因为它不是CheckpointedFunction也没有实现任何快照接口。这意味着作业从 checkpoint 恢复之后生成器的状态是全新的maxTimestamp会被重置到初始值。那会发生什么恢复后的第一批数据会把maxTimestamp重新推上去然后onPeriodicEmit发出一批水位线。由于恢复后的水位线从很小的值开始爬升下游的窗口定时器这些定时器是注册在 keyed state 里的会随 checkpoint 恢复在恢复后要等到水位线重新爬过它们触发点才会执行。如果你的 checkpoint 间隔比较长或者恢复后数据有断档可能出现“恢复后一段时间内所有窗口都不触发然后突然集中触发一大批”的现象。这个行为本身是语义正确的at-least-once 或者 exactly-once 都允许重算但如果你的下游对突发流量敏感需要提前做容量准备。注意如果你确实需要生成器状态参与恢复比如那个自适应乱序估计器恢复后重新学习会有个冷启动期可以自己实现CheckpointedFunction在initializeState里恢复maxTimestamp和桶数组。不过要小心恢复的maxTimestamp如果比当前实际数据的时间戳大很多会导致恢复后立刻把水位线推得很高把新数据全判成迟到。这个权衡得自己拿捏。6.4 关于 final watermark 的一点观察算子关闭的时候默认会补发一条Watermark.MAX_WATERMARK目的是让下游所有还挂着的窗口立刻结算并输出避免作业停止时丢掉最后一批结果。这个行为在不同版本之间有过调整有些版本可以通过参数关掉。我的建议是自己动手验证一下你用的版本到底是什么行为方法是写一个最简单的本地作业一个 source 发几条带时间戳的数据一个滑动窗口做聚合。跑完之后看输出如果最后那批本该被窗口持有的数据被输出出来了说明 final watermark 起了作用。这个验证十分钟就能做完但能帮你避免很多“作业停了结果少了一截”的困惑。另外提醒一句如果你在管道里串了多个assignTimestampsAndWatermarks虽然不推荐但有时为了在不同阶段用不同策略会这么干最终水位线是被最后一个生成器决定的中间那些策略的 final watermark 行为可能互相影响测试的时候要把整条链都跑通。我自己在这个方向上折腾了挺久最大的体会是Watermark 的难点从来不在概念概念三句话就能讲完难点在于它是一条贯穿数据源、算子链、网络传输、状态恢复的完整链路任何一个环节出问题表现都是“窗口不触发”这一个症状。所以排查的时候不要只盯着生成器看要从数据源的时间戳一路往下捋看每个 subtask 的输入水位线、输出水位线、下游的合并结果一级一级对通常十分钟内就能定位。至于参数怎么调别信任何“最佳实践”里给的固定数值那些数字都是在特定数据分布下测出来的你的数据分布不一样答案就不一样老老实实打点、观察、调、再观察这才是唯一靠谱的路子。