企业官网建设流程全解析

1. Watermark 在流处理里到底扮演什么角色

先说结论：Watermark 不是"迟到数据的开关"，而是一句向前推进的承诺——它告诉下游，"时间戳早于我的事件，理论上不会再来（或者来了我也不等）"。这句话说出口，下游才敢关窗口、才敢算结果、才敢把状态清掉。我见过太多人把 watermark 当成"延迟参数"，配了个forBoundedOutOfOrderness(Duration.ofSeconds(5))就以为万事大吉，结果窗口不触发、数据算重、状态只涨不跌，最后半夜爬起来看日志。

这篇内容围绕 Flink 的WatermarkGenerator展开，从接口定义、内置实现、装配链路一直聊到实际调参时踩过的坑。适合已经能跑通基本窗口作业、但还没搞清楚"我这个窗口为什么就是不关"的同学；也适合想读 Flink 源码但被TimestampsAndWatermarksOperator那一坨绕晕的人。我不打算复述官方文档，只讲源码里那些真正决定行为的东西。

1.1 事件时间这条线，其实只有两个角色

把事件时间拆开看，整条链路上只有两个角色在干活：一个是时间戳提取器，负责从每条记录里挖出一个long类型的毫秒值；另一个就是WatermarkGenerator，负责根据这些时间戳决定"水位线推到哪"。前者决定数据的时间坐标，后者决定计算的推进进度。窗口、定时器、状态清理、双流 Join 的对齐，全都建立在这两者的产出之上。

有个容易混淆的点：时间戳提取器只影响"这条记录属于哪个窗口"，完全不影响"什么时候关窗口"。我调试过一次事故，负责的同学把assignTimestampsAndWatermarks写在keyBy之后，导致下游一个算子拿到的水位线和上游完全脱节——窗口在另一个并行度上算的，水位线却是本地算的，结果就是局部窗口早就该触发了，却一直卡着。

DataStream<Event> stream = env.addSource(source) .assignTimestampsAndWatermarks( WatermarkStrategy .<Event>forBoundedOutOfOrderness(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 WatermarkGenerator<T> { 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 BatchEndWatermarkGenerator<T> implements WatermarkGenerator<T> { 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(StreamRecord<T> 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 合成一个。规则大致三条：

它在推进之前会做一次"对齐"检查：只有当水位线大于当前已输出的水位线时才发出去。这就是"只能前进不能后退"的实现位置。有些同学在自定义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 AscendingTimestampsWatermarkGenerator<T> extends BoundedOutOfOrdernessWatermarkGenerator<T> { 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一直停在很早的值上，周期性发出的水位线是个极小的值。多输入算子取最小值，整条链路的水位线就被这几个不活跃分区拖住了。

修复方式就是withIdleness：

WatermarkStrategy .<Event>forBoundedOutOfOrderness(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 这几件事的交汇处。接口两个方法，看懂只要五分钟；但要知道一个参数改动会在多久之后以什么形式反馈到你的指标上，需要真正被坑过几次。我现在的习惯是每次调整水位线相关配置，都会同步改一下监控面板的阈值——因为这个改动的影响往往滞后十几分钟才显现，等肉眼看到的时候，窗口可能已经积了一堆没关的状态了。