Flink 实战：Watermark 的用法与结合 Window 处理延迟数据

• 示例工程
• 大数据

导读

本文是《Flink 实战与性能优化》3.5 节的核心内容，聚焦流式计算中最棘手的两个问题——事件乱序与事件延迟。文章以事件时间（Event Time）为背景，系统讲解 Watermark（水印）的机制原理、Flink 中两种水印分配方式（Punctuated 与 Periodic）的代码实现与适用场景，并结合本仓库源码给出完整可运行的实战示例，最后覆盖 Window 触发后处理迟到数据的三种手段（默认丢弃、allowedLateness、sideOutputLateData）。读完本文，你将掌握在水印机制下正确设计基于事件时间的窗口聚合任务，并针对不同业务容忍度选择合适的迟到数据处理方案。

为什么需要 Watermark：事件乱序与事件延迟

在 3.1 节 中我们讲过 Flink 中的三种时间（Event Time、Ingestion Time、Processing Time）及其使用场景，在 3.2 节中又深入讲解了窗口机制与 Flink 自带 Window 的实现原理。这里需要明确一个关键前提：

• 如果窗口基于Processing Time，Flink 消费数据时完全不需要关心数据本身的时间，因为 Processing Time 代表的是数据在 Flink 中被处理的时间，这个时间是顺序递增的，不存在乱序问题；
• 如果窗口基于Event Time，就必须直面两个问题：事件乱序与事件延迟。

理想情况下，Event Time 与 Process Time 相等，即数据发生的时间与数据处理的时间之间没有延迟。但现实往往"骨感"：网络的抖动、设备的故障、应用的异常等原因，会导致 Process Time 总是滞后于 Event Time 一段距离。

所谓乱序，是指 Flink 接收到的事件的先后顺序，并不是严格按照事件的 Event Time 排列的——先产生的数据可能晚到，后产生的数据反而先到。

依赖事件时间的典型场景

有些场景特别依赖事件发生时间而非处理时间，例如：

• 错误日志分析：错误日志的时间戳代表着错误发生的具体时间，开发者只有知道这个时间戳，才能还原那个时间点系统到底发生了什么问题，或者根据该时间戳去关联其他事件，找出触发问题的根源；
• 设备监控：设备传感器或监控系统按时间点实时上传设备周围监控情况，通过监控大屏实时查看，不错漏重要或可疑的事件。

这类场景下，最有意义的是事件发生的顺序，而不是事件到达 Flink 后被处理的顺序。庆幸的是，Flink 支持用户以事件时间来定义窗口（也支持处理时间），而为了解决乱序与延迟问题，Flink 引入了Watermark 机制。

3.5.1 Watermark 简介：工作原理与触发机制

先看一个业务例子：统计 8:00 ~ 9:00 时间段打开淘宝 App 的用户数量。Flink 可以开一个窗口做聚合，但由于网络抖动、应用采集发送延迟等原因，无法保证在窗口结束那一刻窗口中已经收集齐 8:00 ~ 9:00 内用户打开 App 的所有事件——但又不能无限期等下去。

当基于事件时间的数据流进行窗口计算时，最困难的一点就是：如何确定对应当前窗口的事件已经全部到达。实际上我们并不能百分百准确判断，因此业界常用的做法是：基于已经收集到的消息来估算是否还有消息未到达，这就是 Watermark 的思想。

Watermark 的定义：Watermark 是一种衡量 Event Time 进展的机制，它是数据本身的一个隐藏属性，数据本身携带着对应的 Watermark。Watermark 本质上就是一个时间戳，代表着比这个时间戳更早的事件已经全部到达窗口，即假设不会再有比这时间戳更小的事件到达。

这个假设是触发窗口计算的基础：只有 Watermark 大于窗口对应的结束时间，窗口才会关闭并进行计算。按照这个标准处理数据，如果后面还有比该时间戳更小的数据到达，则被视为迟到的数据——对于这部分迟到数据，Flink 也有相应的机制去处理（下文 3.5.7 节详述）。

Watermark 如何工作：一个 4s 窗口的逐步推演

以 Flink 从消息队列消费数据为例，数据上的数字代表数据本身的 timestamp，W(4)、W(9)代表水印，窗口是基于事件时间定义的 4s 时间窗口：

1. 数据流乱序到达，Flink 消费后，数据1、3、2进入第一个窗口（[0, 4)）；
2. 数据7进入第二个窗口（[4, 8)）；随后乱序的3依旧进入第一个窗口；
3. 接着水印W(4)到达，水印的 timestamp 与第一个窗口结束时间一致，代表后面不会再有比 4 更小的数据到达，于是第一个窗口被触发计算；
4. 后续数据5、6进入第二个窗口，数据9进入第三个窗口；
5. 当水印W(9)到达时，水印比第二个窗口的结束时间8还大，第二个窗口也随之触发计算，以此类推。

整个流程印证了窗口计算的触发条件：Watermark 大于等于窗口 endTime 时，窗口关闭并触发计算。

3.5.2 Flink 中 Watermark 的设置方式

在 Flink 中，数据处理时需要调用 DataStream 的assignTimestampsAndWatermarks方法来分配时间戳与水位线。该方法有两种重载，分别接收AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks，其底层实现如下：

public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks<T> timestampAndWatermarkAssigner) { final int inputParallelism = getTransformation().getParallelism(); final AssignerWithPeriodicWatermarks<T> cleanedAssigner = clean(timestampAndWatermarkAssigner); TimestampsAndPeriodicWatermarksOperator<T> operator = new TimestampsAndPeriodicWatermarksOperator<>(cleanedAssigner); return transform("Timestamps/Watermarks", getTransformation().getOutputType(), operator).setParallelism(inputParallelism); } public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks<T> timestampAndWatermarkAssigner) { final int inputParallelism = getTransformation().getParallelism(); final AssignerWithPunctuatedWatermarks<T> cleanedAssigner = clean(timestampAndWatermarkAssigner); TimestampsAndPunctuatedWatermarksOperator<T> operator = new TimestampsAndPunctuatedWatermarksOperator<>(cleanedAssigner); return transform("Timestamps/Watermarks", getTransformation().getOutputType(), operator).setParallelism(inputParallelism); }

由此，设置 Watermark 有两条路线：

生产环境通常使用周期性生成方式，但 Watermark 的生成方式需要根据业务场景的不同进行不同的选择。

3.5.3 Punctuated Watermark：逐事件判定生成

AssignerWithPunctuatedWatermarks接口包含checkAndGetNextWatermark方法，该方法会在每次extractTimestamp()被调用后调用，由它决定是否生成新的水印。返回的水印只有在不为 null 且时间戳大于先前返回的水印时间戳时才会被发送出去；如果返回 null 或时间戳比之前小，则不生成新的水印。

仓库中提供了完整的自定义实现 WordPunctuatedWatermark.java：

public class WordPunctuatedWatermark implements AssignerWithPunctuatedWatermarks<Word> { @Nullable @Override public Watermark checkAndGetNextWatermark(Word lastElement, long extractedTimestamp) { return extractedTimestamp % 3 == 0 ? new Watermark(extractedTimestamp) : null; } @Override public long extractTimestamp(Word element, long previousElementTimestamp) { return element.getTimestamp(); } }

该实现以extractedTimestamp % 3 == 0作为生成水印的判定条件：时间戳能被 3 整除的事件到达时生成一个水印，其余事件不生成。其配套数据模型 Word.java 只包含word、count、timestamp三个字段，时间戳直接从事件中提取。

注意：这种方式理论上可以为每个事件都生成一个水印，但水印要参与下游计算，水印过多会导致整体计算性能下降，因此只适合对实时性要求极高的场景。

配套的可运行示例见 Main.java，其中通过env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)声明使用事件时间，并通过env.socketTextStream("localhost", 9001)读取 socket 数据，Map 解析为 Word 后调用assignTimestampsAndWatermarks(new WordPunctuatedWatermark())。

3.5.4 Periodic Watermark：周期性生成

生产环境中使用AssignerWithPeriodicWatermarks定期分配时间戳并生成水印更为普遍。仓库中的 WordPeriodicWatermark.java 给出了完整实现（比书中示例多加了日志输出，便于观察水印推进过程）：

@Slf4j public class WordPeriodicWatermark implements AssignerWithPeriodicWatermarks<Word> { private long currentTimestamp = Long.MIN_VALUE; @Override public long extractTimestamp(Word word, long previousElementTimestamp) { long timestamp = word.getTimestamp(); currentTimestamp = Math.max(timestamp, currentTimestamp); log.info("event timestamp = {}, {}, CurrentWatermark = {}, {}", word.getTimestamp(), DateUtil.format(word.getTimestamp(), YYYY_MM_DD_HH_MM_SS), getCurrentWatermark().getTimestamp(), DateUtil.format(getCurrentWatermark().getTimestamp(), YYYY_MM_DD_HH_MM_SS)); return word.getTimestamp(); } @Nullable @Override public Watermark getCurrentWatermark() { long maxTimeLag = 5000; return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - maxTimeLag); } }

该类实现两个方法：

• extractTimestamp()：从数据中提取 Event Time，并将当前时间戳与事件时间比较取最大值后赋给currentTimestamp，最后返回事件时间；
• getCurrentWatermark()：通过currentTimestamp - maxTimeLag得到水印值。其中maxTimeLag代表数据允许延迟的时间，示例中long maxTimeLag = 5000;表示最大允许数据延迟 5 秒。超过 5 秒之后如果还来了更早的数据，Flink 会将其丢弃——因为窗口中的数据需要被触发，不可能一直等待迟到的数据（例如因网络问题迟迟未上传的数据）而不结束计算。

合理设置允许延迟时间是一门细活：需要观察生产环境从数据采集、进入消息队列再到 Flink 的整个流程是否出现延迟，统计平均延迟的大致波动范围。这也说明一个事实：Flink 设计 Watermark 的根本目的是解决部分数据乱序或延迟问题，但不能真正做到彻底解决——不过在流处理框架中这已经是非常实用的特性了。

Periodic 的四个内置实现类

AssignerWithPeriodicWatermarks接口有四个实现类，功能与使用方式如下：

1. BoundedOutOfOrdernessTimestampExtractor

用于发出滞后于数据时间的水印，作用与上面自定义的类类似，只需传入一个时间参数代表允许数据延迟到来的时间。使用方式（仓库示例 Main2.java）：

// Time.seconds(10) 代表允许延迟的时间大小 data.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Word>(Time.seconds(10)) { // 重写 extractTimestamp() 抽象方法 @Override public long extractTimestamp(Word element) { return element.getTimestamp(); } });

2. CustomWatermarkExtractor

仓库中自定义的周期性生成水印类（MetricWatermark.java），用于对MetricEvent等数据流生成水印，配合 flink-learning-common 模块的公共模型使用。

3. AscendingTimestampExtractor

用于时间戳单调递增的数据流。如果数据流的时间戳不是单调递增，会有专门的处理方法，其核心逻辑为：

public final long extractTimestamp(T element, long elementPrevTimestamp) { final long newTimestamp = extractAscendingTimestamp(element); if (newTimestamp >= this.currentTimestamp) { this.currentTimestamp = newTimestamp; return newTimestamp; } else { violationHandler.handleViolation(newTimestamp, this.currentTimestamp); return newTimestamp; } }

即新时间戳不小于当前时间戳时正常推进；否则触发violationHandler.handleViolation(...)处理乱序违规。

4. IngestionTimeExtractor

依赖机器系统时间：在extractTimestamp和getCurrentWatermark方法中基于System.currentTimeMillis()获取时间，而不是基于事件的时间。如果该分配器在数据进入 Flink 后立即分配，这个时间就与 Ingestion Time 一致，因此得名 IngestionTimeExtractor。

两个重要的使用注意点

1. 使用周期性方式生成水印时，可以通过env.getConfig().setAutoWatermarkInterval(...)设置水印生成间隔（每隔 n 毫秒）。仓库示例 Main1.java 中设置了env.getConfig().setAutoWatermarkInterval(5000);表示每 5 秒生成一次水印。

2. 通常建议在数据源（source）之后就生成水印，或者先做 filter/map/flatMap 等简单操作之后再生成水印——越早生成水印效果越好，甚至可以直接在数据源头生成。例如在 source 的run()方法中：

@Override public void run(SourceContext<MyType> ctx) throws Exception { while (/* condition */) { MyType next = getNext(); ctx.collectWithTimestamp(next, next.getEventTimestamp()); if (next.hasWatermarkTime()) { ctx.emitWatermark(new Watermark(next.getWatermarkTime())); } } }

通过ctx.collectWithTimestamp携带事件时间输出，通过ctx.emitWatermark主动发射水印。

3.5.5 每个 Kafka 分区的时间戳

当使用 Kafka Connector 作为数据源时，Flink 的 Kafka Consumer 会按分区读取数据，并为每个 Kafka 分区分别分配时间戳与水印。Kafka 单分区内的消息通常是有序的，因此在分区级别生成的水印质量更高；当多个分区的水印汇聚到下游算子时，Flink 取所有输入分区水印的最小值作为该算子的当前水印，以确保不会漏掉任何分区中可能迟到的数据。这一机制可以在使用 Kafka 作为数据源的作业中通过assignTimestampsAndWatermarks覆盖 Kafka Consumer 默认的时间戳提取与水印生成逻辑（参考仓库中 Kafka 相关连接器示例）。

3.5.6 将 Watermark 与 Window 结合起来处理延迟数据

将水印与窗口结合，是处理乱序、延迟数据的标准做法：窗口基于 Event Time 定义，窗口的触发由 Watermark 驱动。仓库中的窗口模块 flink-learning-window 也大量采用这一组合。

仓库示例 Main3.java 展示了水印 + 时间窗口 + 迟到容忍的完整链路：

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.setParallelism(1); SingleOutputStreamOperator<Word> data = env.socketTextStream("localhost", 9001) .map(new MapFunction<String, Word>() { @Override public Word map(String value) throws Exception { String[] split = value.split(","); return new Word(split[0], Integer.valueOf(split[1]), Long.valueOf(split[2])); } }).assignTimestampsAndWatermarks(new WordPeriodicWatermark()); data.keyBy(0) .timeWindow(Time.seconds(10)) .allowedLateness(Time.milliseconds(2)) .sum(1) .print(); env.execute("watermark demo");

整个流程为：socket 读入形如word,count,timestamp的文本 → 解析为 Word 并分配时间戳/水印（允许 5s 延迟）→ 按第 0 个字段分组 → 10 秒滚动窗口聚合 count → 窗口触发后额外等待 2ms 的迟到数据。

完整的运行环境依赖：这些示例需要本地启动nc -lk 9001之类的 socket 服务端发送数据；工程位于 flink-learning-examples，其依赖关系可查看模块 pom.xml。

3.5.7 处理延迟数据的三种方法

当窗口已经触发计算后，仍可能有迟到数据到达，Flink 提供三种处理手段：

丢弃（默认）

默认行为。窗口触发计算后，再到达的迟到数据（Watermark 已越过窗口 endTime、且超出 allowedLateness 容忍范围的数据）直接丢弃，不参与计算，也不可恢复。这种方式最简单、开销最低，适用于对结果精度要求不高的场景。

allowedLateness：再次指定允许数据延迟的时间

在窗口上调用allowedLateness(Time)可以额外指定一段允许数据延迟的时间。在窗口已经因 Watermark 触发后，只要迟到的数据仍未超过窗口结束时间 + allowedLateness，该数据依然会被重新纳入对应窗口，并触发窗口的再次计算（输出更新后的聚合结果）。

仓库示例 Main3.java 中：

data.keyBy(0) .timeWindow(Time.seconds(10)) .allowedLateness(Time.milliseconds(2)) .sum(1) .print();

即 10 秒窗口触发后，再等待 2ms 内的迟到数据仍会参与计算。

使用注意：

• allowedLateness与 Watermark 中的maxTimeLag是两个不同的"延迟容忍"：前者针对窗口触发之后的迟到数据，后者针对水印推进本身的滞后；
• 过大的allowedLateness会让窗口迟迟无法真正清理状态、增加状态存储压力，并导致结果反复更新，需结合实际业务权衡。

sideOutputLateData：收集迟到的数据

对于超出允许范围、即将被丢弃的迟到数据，可以通过sideOutputLateData(OutputTag)将其路由到旁路输出流中，方便后续单独存储、修复或重新计算，做到迟到数据"不丢失、可追溯"。

仓库示例 Main4.java 完整展示了用法：

OutputTag<Word> lateDataTag = new OutputTag<Word>("late") { }; SingleOutputStreamOperator<Word> data = env.socketTextStream("localhost", 9001) .map(new MapFunction<String, Word>() { @Override public Word map(String value) throws Exception { String[] split = value.split(","); return new Word(split[0], Integer.valueOf(split[1]), Long.valueOf(split[2])); } }).assignTimestampsAndWatermarks(new WordPeriodicWatermark()); SingleOutputStreamOperator<Word> sum = data.keyBy(0) .timeWindow(Time.seconds(10)) // .allowedLateness(Time.milliseconds(2)) .sideOutputLateData(lateDataTag) .sum(1); sum.print(); sum.getSideOutput(lateDataTag) .print(); env.execute("watermark demo");

要点拆解：

• 先定义一个匿名的OutputTag<Word>（new OutputTag<Word>("late") {}），泛型与数据类型一致；
• 在窗口算子链上调用.sideOutputLateData(lateDataTag)声明迟到数据输出到该 Tag；
• 主结果流通过sum.print()打印正常聚合结果；
• 迟到数据通过sum.getSideOutput(lateDataTag).print()从旁路输出流中取出打印，实现"主结果与迟到数据分离消费"。

这三种方式可以组合使用：例如既设置allowedLateness容忍一小段窗口触发后的数据，又用sideOutputLateData把更晚的数据保存下来做离线补救。

3.5.8 小结与反思

回顾本节，核心脉络可以概括为：

1. 问题来源：选择 Event Time 就必须面对事件乱序与事件延迟；而 Processing Time 天然顺序递增，不需要 Watermark。
2. Watermark 本质：一种衡量 Event Time 进展的机制，是数据携带的隐藏时间戳属性，代表"此时间戳之前的事件均已到达"的假设，是触发窗口计算的依据。
3. 两种分配方式：Punctuated（逐事件判定、实时性高、水印量大）与 Periodic（周期性生成、生产主流），均可通过assignTimestampsAndWatermarks接入，且需配合setAutoWatermarkInterval控制生成频率；越早生成水印效果越好，甚至可在 source 源头通过emitWatermark直接发射。
4. 内置工具：BoundedOutOfOrdernessTimestampExtractor（允许固定延迟）、AscendingTimestampExtractor（单调递增流）、IngestionTimeExtractor（依赖机器时钟）覆盖了常见场景；Kafka 场景下可按分区分配时间戳，汇聚时取最小水印。
5. 迟到数据三板斧：默认丢弃、allowedLateness二次容忍、sideOutputLateData旁路收集，三者结合可灵活适配从"粗粒度近似"到"精确可回溯"的不同需求层次。

需要反思的是：Watermark 只能缓解、无法根治乱序与延迟。maxTimeLag、allowedLateness等参数都依赖对生产链路（采集 → 消息队列 → Flink）延迟分布的持续观测与调优，脱离业务数据特征盲目设参，要么造成结果不准，要么造成状态膨胀。建议在实际项目中先基于监控数据统计延迟分布，再据此设定水印滞后与迟到容忍参数，并在上线后持续观察指标进行调整。

本节完整配套源码位于仓库 flink-learning-examples 的streaming/watermark包下：Word 数据模型、两种水印生成器（WordPeriodicWatermark、WordPunctuatedWatermark）以及 Main ~ Main4 五个可运行示例，均可直接编译运行验证本文所述机制。

• 示例工程
• 大数据