企业官网建设流程全解析

Flink 事件时间这套语义里，Watermark（水位线）是绕不过去的一环。它看起来只是一个随数据流走的 Long 值，但它决定了窗口什么时候触发、状态什么时候能清、结果什么时候能吐出来，也决定了你在半夜被“指标不动了”的告警叫起来时，知道该往哪个方向查。这篇内容围绕 Flink 事件时间与 Watermark 展开，把定义、生成方式、传播对齐、窗口触发公式、允许延迟与侧输出的配合、监控指标和参数估算讲透，代码部分给出可以直接改成自己项目能跑的最小版本。适合已经写过几个 Flink 作业、但碰到乱序数据或者“窗口迟迟不出结果”这类问题时心里没底的同学，也适合刚接手一个流式作业、需要快速建立整体认知的读者。我尽量不写教科书式的推导，而是把我在实际项目里踩到的坑和判断依据摊开说。

1. 先把 Watermark 要解决的问题说清楚

1.1 三种时间语义，选的不是口味而是代价

流处理里给数据打时间戳，可选的基准无非三种：处理时间、摄入时间、事件时间。处理时间取的是机器当前时钟，实现最简单，延迟最低，但同一条数据重跑两次算出来的结果可能完全不同，因为它绑定的是“算的时候是几点”。摄入时间取的是数据进入源算子的时刻，比处理时间稳定一些，本质仍然依赖运行环境。事件时间取的是数据自己携带的业务时间，比如订单创建时刻、设备上报时刻、日志里那条记录的产生时刻，它唯一的好处同时也是最大的代价：结果可重放、可回溯、口径与业务对齐，但你必须自己处理“数据到得不整齐”这件事。

我见过不少项目一开始图省事用处理时间，业务方看到报表里某个小时的数字第二次刷新时变了，就会来问为什么。这类问题最后基本都要回到事件时间。代价也很明确：一是要求上游数据必须带可靠的时间戳字段，二是你要接受“结果会等”，也就是窗口不再是一到点就出，而是要等水位线越过窗口边界。这个等待到底等多久，就是 Watermark 要回答的问题。

所以判断标准很直接：如果业务对同一份输入要求结果确定、可重算、口径与业务时间对齐，就必须走事件时间，然后老老实实把 Watermark 配好。如果只是做监控告警、对结果一致性没有严格要求、又不能容忍延迟，处理时间反而更合适。不要因为“事件时间是高级特性”就无脑上，我见过把处理时间作业硬改成事件时间，结果延迟涨了十几秒，业务方完全不接受的案例。

1.2 乱序到底乱在哪：用一条订单数据看清楚

假设你有一条订单流，每条记录带一个 eventTime。上游是一个多分区的消息队列，同一个用户的两次下单可能被投递到不同分区，消费者并行拉取，谁先到算子完全说不准。于是你会看到算子收到的序列大概是这样的：10:00:05 的记录，接着 10:00:03，接着 10:00:08，接着 10:00:02。这不是异常，这是分布式系统里的常态，来源包括分区之间消费速度不均、上游生产端重试、网络抖动、中间还有一层聚合服务做了缓冲。

如果没有水位线，你面对的问题就是：10:00:00 到 10:00:10 这个窗口，我什么时候可以算？来一条算一条？那结果就是错的。等到“确定没有 10:00:09 的数据了”再算？那你怎么知道没有了？系统并没有一个全局的“数据到齐”信号。Watermark 就是人为造出来的这个信号，它是一句承诺：事件时间小于等于这条水位线的数据，我认为都到齐了。

这里必须强调“认为”两个字。它不是一个事实，是一个基于乱序容忍度做出的假设。你把容忍度设成 5 秒，就意味着你相信“任何一条数据，最多迟到 5 秒”。如果现实打脸，迟到 8 秒的数据进来，那它就会被当成迟到数据，要么被丢掉，要么走侧输出，要么在允许延迟的窗口里被重新计算。所以 Watermark 的配置本质是一次押注，押的是你对上游乱序分布的了解程度。

1.3 定义与那条不能破的单调红线

严格的表述是：watermark 为一个时间戳 T 的水位线，表示事件时间小于等于 T 的数据已经全部到达，后续不应该再出现时间戳小于等于 T 的数据。对应到窗口上，一个区间为 [start, end) 的窗口，当水位线满足 watermark >= end 时触发计算。也就是说水位线是“窗口结束时间”这把锁的钥匙，越过了就开锁。

由此推出一个硬约束：水位线必须单调不减。这条线只能往前推，不能往后退，否则已经触发过的窗口会被重复触发，语义就全乱了。Flink 在实现上强制了这一点，你在自定义生成器里返回一个更小的值也不会生效，因为内部取的是当前值与新值的较大者。

另一个容易忽略的点是水位线是“跟着数据流走”的。它不是某个全局变量，而是作为一类特殊记录插在数据流中间，随 record 一起在算子间传递。这意味着它受反压影响，也受分区影响。当你在 UI 上看到某个算子的输入水位线卡住不动，第一反应应该是往上游看：是没数据了，还是上游被反压堵住了，还是某个分区空转导致对齐卡住了。

注意：水位线卡住不等于作业挂了。数据流停更、某个分区长时间无数据、上游算子反压，这三种情况在下游看到的现象完全一样，必须结合 source 端的消费速率和反压指标一起判断。

2. Watermark 的生成方式与参数拆解

2.1 forBoundedOutOfOrderness 里那 1 毫秒的意义

绝大多数场景用的是内置的WatermarkStrategy.forBoundedOutOfOrderness(Duration)。它的生成逻辑用一句话概括：记录下目前为止见过的最大事件时间maxTimestamp，产出的水位线等于maxTimestamp - outOfOrderness - 1ms。那个减 1 毫秒经常被忽略，但它很关键。

举个具体数字。窗口是 [10:00:00, 10:00:10)，结束时间 end = 10:00:10，触发条件是 watermark >= 10:00:10。容忍度设 5 秒。如果水位线公式是maxTimestamp - 5s，那么当最大时间戳到达 10:00:15 时，水位线正好等于 10:00:10，窗口触发。问题在于，此时刚刚到达的这条 10:00:15 的数据，本身是否属于这个窗口？不属于。但如果水位线没有减 1 毫秒，一条刚好落在 10:00:15 整的数据会让水位线等于 10:00:10，理论上还有可能来一条 10:00:10 之前的数据，那它就迟到了。减掉 1 毫秒，让水位线在最大时间戳达到 10:00:15 加一点点时才越过边界，等价于把“小于等于水位线的都到齐”这个承诺收紧了一毫秒，边界上不会漏。这个细节在官方的测试用例和大量博客里都有提及，理解它之后你就不会写出“容忍度 5 秒但边界数据总被丢”的诡异现象。

代码上就一行：

WatermarkStrategy<OrderEvent> strategy = WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, recordTs) -> event.getEventTime()) .withIdleness(Duration.ofSeconds(30));

withTimestampAssigner里第二个参数是记录进入系统时的处理时间戳，一般用不上，尽量不要拿它兜底，否则时间语义就混了。时间戳的单位是毫秒，如果你上游给的是秒级时间戳或者字符串，务必在 assigner 里转换干净，我见过把秒当毫秒传的，水位线直接跑到 1970 年，窗口永远不触发。

2.2 什么时候必须自己写周期性生成器

内置的 fixed-delay 策略对“全局统一容忍度”这个假设要求很高。如果上游不同分区、不同业务类型的乱序程度差异很大，一个全局值就会出问题：设大了整体延迟高、状态大；设小了某类数据全被判定迟到。这时候需要自定义PeriodicWatermarkGenerator：对每条记录维护一个“业务上合理的可接受时间戳”，可以是基于事件的某个字段（比如上游给的采集完成时间），也可以是基于数量的，比如每积压 1000 条推进一次。

public class AdaptiveWatermarkGenerator implements WatermarkGenerator<OrderEvent> { private final long outOfOrdernessMillis = 5000L; private long maxTimestamp = Long.MIN_VALUE; @Override public void onEvent(OrderEvent event, long eventTimestamp, WatermarkOutput output) { maxTimestamp = Math.max(maxTimestamp, eventTimestamp); } @Override public void onPeriodicEmit(WatermarkOutput output) { if (maxTimestamp == Long.MIN_VALUE) { return; // 还没数据，不要发 watermark } output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1)); } }

这类生成器的价值在于你可以把“容忍度”做成动态的：根据上游上报的采集延迟监控，或者根据最近一段时间迟到数据的分位数，定时调整outOfOrdernessMillis。注意调整的幅度要缓，因为它会直接影响延迟。我在一个项目里做过基于迟到分布 P99 的动态值，实现上就是一个滑动统计窗口，代价是代码复杂度上升了不少，如果上游稳定性足够，老老实实写死一个值更省心。

2.3 Punctuated 生成的边界与隐患

PunctuatedWatermarkGenerator是“数据驱动”的：每来一条记录，你自己决定要不要发出水位线。典型用法是遇到某种标记事件（比如一批数据的结束标记、心跳、EOF）时推进水位线。

它的优点非常诱人：延迟低，不用等周期性触发那 200 毫秒；语义清晰，标记到了就推进。但隐患也很实在。第一，它要求数据流里真的存在这种标记，很多场景根本不存在。第二，如果每条记录都发水位线，下游每个分区都会频繁处理水位线事件，网络和 CPU 开销上去得很快，还容易形成大量小消息。第三，也是最容易出事的：如果标记事件稀疏，水位线推进完全依赖它，一旦某个分区长时间没有标记，全局水位线就被拖住。

我的建议是把它当成一种“有明确业务分隔符”时的补充手段，比如离线补数和实时流混跑，用标记区分批次。纯实时流里，周期性生成加足够合理的容忍度，稳定性要好得多。

2.4 并行流里的 Watermark 对齐：为什么是取最小

这是新手最容易懵的地方。一个算子的并行实例可能有多个输入通道（来自上游多个并行实例），每个通道都在推自己的水位线。算子对外发出的水位线，取的是所有输入通道水位线的最小值。逻辑很直白：只有所有输入都保证“这个时间点之前的数据到齐了”，算子才能说自己也到齐了。

假设上游并行度是 4，其中三个分区已经推到了 10:00:30，第四个分区因为分配的数据少、或者消费慢，还停在 10:00:05。那这个窗口算子对外发出的水位线就是 10:00:05，所有窗口都得等。这就是“木桶效应”。如果你在 UI 上看到某个算子的输出水位线长期不涨，但有些上游实例的输入水位线涨得很快，那基本就是某个通道拖后腿了。

提示：并行度越高、分区之间数据倾斜越明显，这个问题越容易暴露。keyBy 之后的窗口算子通常输入通道多，特别容易撞上；如果上游是全量广播或者并集，情况会更复杂。

这也解释了一个实践原则：时间戳分配和 Watermark 生成尽量放在离 source 最近的地方，最好直接在 source 上声明。很多版本的连接器支持在 source builder 上直接设置WatermarkStrategy，这样水位线在源头就注入，不会因为中间某个算子并行度变化导致传播路径变绕。如果你在 keyBy 之后才分配时间戳，那水位线的传播会跟着分区走，出问题的概率明显变大。

3. 一套可以照着跑的实操配置

3.1 数据源与时间戳分配的位置选择

先给一个能跑的最小骨架。数据源用消息队列的并行消费，模拟 8 个分区，每个分区内部有序、分区之间有乱序。时间戳就在 source 之后立刻分配，不做任何中间转换。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(4); env.getConfig().setAutoWatermarkInterval(200L); // 默认就是 200ms，显式写出来便于排查 DataStream<OrderEvent> source = env .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "order-source") .uid("order-source"); DataStream<OrderEvent> timestamped = source.assignTimestampsAndWatermarks( WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((e, ts) -> e.getEventTime()) .withIdleness(Duration.ofSeconds(30)) );

这里有三个决策点值得说。第一，source 上用了noWatermarks()，因为时间戳还没解析，让 source 先别生成水位线，避免产生一堆无意义的 -Long.MAX_VALUE。第二，setAutoWatermarkInterval(200)是周期性生成器的发射间隔，它决定了水位线推进的“粒度”，也决定了最小延迟的下限之一，不要设得太小，几十毫秒级别意义不大还费 CPU。第三，withIdleness(30s)是给空闲分区兜底的，理由在 3.3 节展开。

3.2 窗口、允许延迟、侧输出三件套的组合

窗口触发只解决了“主逻辑什么时候算”，迟到数据怎么处理是另一半。三个参数必须一起看：容忍度outOfOrderness、窗口允许延迟allowedLateness、侧输出sideOutputLateData。

OutputTag<OrderEvent> lateTag = new OutputTag<OrderEvent>("late-orders") {}; SingleOutputStreamOperator<OrderStat> result = timestamped .keyBy(OrderEvent::getMerchantId) .window(TumblingEventTimeWindows.of(Time.seconds(10))) .allowedLateness(Time.seconds(30)) .sideOutputLateData(lateTag) .aggregate(new OrderAggregate(), new OrderWindowFunction()); DataStream<OrderEvent> tooLate = result.getSideOutput(lateTag);

三者的分工是这样：容忍度管的是“水位线推到哪”，它是全局的、一次性的判断；允许延迟管的是“窗口触发之后还能不能收数据”，窗口状态会保留到end + allowedLateness才销毁；侧输出管的是“连允许延迟都过了的数据去哪”。也就是说，一条数据的命运有三种：准时进入窗口计算；在水位线之后、允许延迟之内到达，触发窗口重新计算并输出更新结果；超过允许延迟，进侧输出。

这里有个必须记住的语义陷阱：允许延迟导致的重新计算，会让下游收到同一个窗口的多条结果。下游如果是直接写数据库，必须能处理覆盖或者带版本；如果是写消息队列，消费方要么幂等，要么按窗口标识去重。我见过一个作业开了 30 秒允许延迟，下游没做幂等，结果报表里同一个窗口的数字被累加了五次，查了整整半天。

3.3 空闲分区与那个永远不来的最后一次触发

空闲分区的问题值得单独拎出来讲，因为它太常见了。假设上游有 16 个分区，某个时段只有 3 个分区有数据，另外 13 个分区没有任何新消息。这些分区的水位线停在最后一个数据的时间点不再前进，而算子取最小值，于是全局水位线被死死拖住，窗口一个都不触发，表现为“作业没报错、数据也在进、但下游一条结果都收不到”。

解决办法就是withIdleness(Duration)：当一个分区超过设定时长没有任何数据，就把它标记为空闲，在计算全局水位线时忽略它，等新数据来了再恢复。30 秒这个值要结合你的数据稀疏程度定，设太小容易误判（分区只是短暂没数据就被忽略，之后来的数据会变成迟到数据），设太大又救不了场。我的经验是取“正常数据间隔的 3 到 5 倍”。

还有一个更隐蔽的坑：整条流都停了。比如业务低谷期，某个商户一个小时没有任何订单，那这个商户对应的最后一个窗口永远等不到水位线越过边界，结果就是不会输出。这个不是 bug，是事件时间语义的必然。要解决只能靠辅助手段：加一个基于处理时间的定时触发，或者在业务侧接受“无数据即无结果”，由下游补零。

注意：withIdleness要加在 WatermarkStrategy 上，加在窗口上没用。有些同学在 SQL 里写作业，对应的参数是table.exec.source.idle-timeout，位置和 DataStream API 不一样，别配错地方。

3.4 超大并行度下的水位线对齐

当 source 分区数上千、并行度很高时，会出现另一个极端：某些分区水位线跑得飞快，某些慢悠悠。快的那部分，数据早就处理完并被缓存进窗口状态，但因为全局取最小值，这些状态一直没法释放，状态体积飙升，checkpoint 越来越大，最后反压。

新版本提供了 source 端的水位线对齐能力，思路是给每个分区的推进速度设一个上限：如果某个分区的水位线超出全局最慢分区太多，就暂停读取它，等落后的追上来。相关配置通常在作业级开启，并设置一个对齐间隔。开启之后的效果是状态体积明显收敛，代价是快分区会被主动限速，整体吞吐可能略降，同时作业对上游反压的敏感度变化需要重新观察。

我的取舍经验是：分区数少于几百、倾斜不严重，不必开，收益有限还多一层复杂度；分区数上千、状态大、checkpoint 时间成为瓶颈，值得试，但要配合观察吞吐和延迟指标，别只看 checkpoint 大小一个数字。

4. 线上排障：水位线不推进的几种典型现场

4.1 从指标和界面定位卡点

排查顺序我一般固定成三步。第一步看 source 的消费速率和是否有积压，确认数据到底在不在进。第二步看算子级的输入、输出水位线指标，找第一个“输入在涨、输出不涨”的算子，那就是卡点所在。第三步再看这个算子的输入通道分布，判断是哪个上游通道拖住了最小值。

Flink 的 Web UI 里，算子详情页能看到当前输入水位线和输出水位线的数值，指标系统里对应的是水位线相关的 gauge。观察技巧是：不要只看一个时间点的值，要看趋势。正常推进的水位线应该是锯齿状上升，周期性发射那 200 毫秒一个台阶。如果是一条水平直线，说明完全不推；如果是缓慢爬升但远慢于数据实际时间，说明容忍度设太大或者某个通道拖后腿。

还有一个信号很有用：迟到数据的计数。如果你接了侧输出，直接统计侧输出流的数据量；如果没有，看窗口的重复触发频率。侧输出量突然暴涨，通常意味着上游出现了短时的大延迟，或者容忍度被设小了。

4.2 常见问题速查表

表格里每一条我都至少遇到过一次，其中“时间戳单位错误”是最容易被忽略的。它不报错，只是水位线跑到一个荒诞的值，窗口安静地不触发。上线前抽样打印几条记录的时间戳，能省掉很多排查时间。

4.3 参数估算：容忍度到底设多少

这个问题没有标准答案，但有可操作的估算路径。第一步，取一段真实数据，统计每条记录的处理时间戳与事件时间戳之差，也就是端到端延迟的分布。第二步，看这个分布的分位数，通常关心 P99 和 P99.9。第三步，把容忍度设成 P99 对应的值，再往上留一点余量。

举个量化的例子。统计出延迟 P99 是 3.2 秒，P99.9 是 7.8 秒。如果容忍度设 4 秒，那么大约 1% 的数据会走迟到路径；如果设 8 秒，迟到数据基本消失，但整体结果延迟增加约 4 秒，窗口状态量也随之增加。选哪个取决于业务能忍多少延迟、以及迟到数据重要不重要。订单金额统计通常不能容忍丢数据，宁可延迟大一点；而实时大屏这种场景，迟到数据占比小的时候直接侧输出丢掉也能接受。

状态量的估算也要一起算。窗口状态大致正比于“窗口长度 + 允许延迟”这段时间内落进来的数据量，与容忍度间接相关：容忍度越大，水位线越滞后，同时在内存里活着的窗口越多。假设每秒 1 万条、每条 200 字节、窗口 10 秒、容忍度 5 秒、允许延迟 30 秒，那水位线滞后加上窗口存活期大约是 45 秒，状态基本就是这个时间窗内的全量数据，粗算一亿字节量级，接近 100MB，再乘上 key 的分布不均和序列化开销，实际会更大。这个粗算能帮你判断该不该调小容忍度，或者改用增量聚合代替全量缓存。

5. 和 Watermark 强耦合的几个细节

5.1 状态清理、允许延迟与 TTL 的边界

很多人以为窗口一触发，状态就释放了。只有在没开允许延迟的情况下才成立。开了允许延迟，窗口状态要保留到windowEnd + allowedLateness之后，因为期间来的迟到数据还要触发重算。这段保留期是实打实的内存占用，允许延迟设 30 分钟和设 30 秒完全不是一个量级。

状态 TTL 是另一套机制，它按“最后访问时间 + TTL”清理，粒度是状态条目。两套机制不要混着用：如果你既开了允许延迟，又配了很短的 TTL，可能出现窗口状态被 TTL 提前清掉、迟到的数据进来找不到窗口而悄悄走侧输出的情况。我的做法是窗口场景下不开 TTL，让窗口生命周期自己管理状态，把复杂度控制在一个地方。

另外要留意聚合方式。用aggregate或reduce做增量聚合，窗口里只存累加器，状态很小；用process全量缓存元素，状态就是元素本身乘以数量。同样一个窗口，两种写法状态量差几个数量级。水位线滞后本身不直接吃内存，真正吃内存的是“滞后期间活着的窗口里缓存的东西”。

5.2 多流 join 与 union 下的水位线传播

双流 join 是水位线最考验人的地方。两个输入流各自生成水位线，join 算子取两者的最小值作为自己的水位线。如果两条流的流量差异很大，比如一条是主订单流，一条是支付回调流，回调流偶尔才来一条，那它的水位线长期不推进，join 算子就被拖住，主订单流的数据全部堆在状态里等匹配。

处理手段有几个：给稀疏的那条流设更激进的容忍度（甚至接近 0），让它尽快推进；给两条流都加withIdleness；在业务上改成 interval join 或者把稀疏流做成维表广播。如果是union而不是 join，水位线同样取最小值，逻辑一致，问题也一样。

还有一种情况是两个流的时间戳来源不一致，一个是毫秒一个是秒，或者一个是本地时间一个是 UTC。join 上之后表现为永远匹配不上或者全部匹配上，前者因为水位线永远不越界，后者因为时间区间完全重叠。上线前把两条流各抽样打几条时间戳，是最省事的验证。

5.3 与 checkpoint 和反压的相互影响

水位线是流里的特殊记录，它和 checkpoint 的 barrier 一样在算子间流动。这意味着当算子发生 barrier 对齐等待时，水位线也跟着停。反压时也一样：下游处理不过来，上游的发送缓冲满了，水位线卡在缓冲区里，于是窗口迟迟不触发。所以我排查“窗口不出结果”时，一定会先看一眼反压指标和端到端延迟，而不是直接怀疑水位线逻辑。

反过来，水位线也会影响 checkpoint。容忍度大、水位线滞后、状态不释放，导致 checkpoint 变大变慢；checkpoint 变慢，又会让反压更严重。这两者容易形成正反馈，最后表现为作业整体卡死。遇到这种情况，先收敛容忍度让状态降下来，比直接加资源更有效。

5.4 SQL 与 Table API 里的声明差异

SQL 里声明水位线是独立语法，跟窗口定义分开写：

CREATE TABLE orders ( merchant_id STRING, amount DECIMAL(10, 2), event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'orders', 'properties.bootstrap.servers' = '...', 'format' = 'json' );

注意这里的INTERVAL '5' SECOND和内建的 fixed-delay 语义是一致的，同样带有边界处理。常见踩坑点是时间字段类型：用TIMESTAMP(3)和用BIGINT存毫秒，水位线的行为不一样。用 BIGINT 的话需要在查询里显式转换，而且容易忘。另外 SQL 里迟到数据的处理方式和 DataStream API 不同，通常靠table.exec.source.idle-timeout处理空闲分区，靠作业参数控制状态保留，没有侧输出那么直接的手段，需要提前规划好数据兜底路径。

还有一个细节：SQL 里如果某个源表没有声明水位线，但又参与了基于事件时间的窗口或 join，优化器会用处理时间语义或者直接报错，取决于版本。这类问题在本地小数据量测试时经常看不出来，上了生产才发现行为不对，所以建表语句里的水位线声明要当成接口契约来对待，别省略。

我在实际项目里的体会是，水位线的参数不值得追求“调优到极致”。绝大多数作业，容忍度设成迟到分布 P99 就够了，剩下的精力放在三件事上：时间戳分配的位置尽量靠前、空闲分区一定要配、下游对重复输出必须有幂等处理。这三件做到位，因为水位线引起的线上问题能少掉八九成。至于再往深了走，比如动态容忍度、基于迟到分布的反馈调节，等你真的被状态体积逼到墙角了再考虑也不迟，过早优化只会让代码变得难维护。