java

4.2 Flink 时间语义与 Watermark

Flink 时间语义深度解析：EventTime 分配 TimestampAssigner、Watermark 生成策略 forBoundedOutOfOrderness/forMonotonousTimestamps、Watermark 传播与空闲检测 idle timeout、AllowedLateness 与 SideOutput 处理迟到数据。

Leeting Yan 2026-11-01 3 分钟阅读 1190 字

为什么需要 Watermark

分布式系统中，事件从产生到被处理存在乱序：

实际发生顺序：A(06:00:00) → B(06:00:01) → C(06:00:02)
到达 Flink 顺序：B(06:00:01) → A(06:00:00) → C(06:00:02)
                              ↑ A 迟到了 1 秒

如果使用 Processing Time，窗口按到达时间分配，A 会被错误分配到后面的窗口。

Event Time + Watermark 解决这个问题：

Event Time：事件实际发生时间（业务语义）
Watermark：推断「不会再有更早数据到达」的时间标记

Watermark 工作原理

数据流：B(06:00:01) → A(06:00:00) → C(06:00:02) → D(06:00:05) → ...
Watermark：                    W(06:00:00)        W(06:00:02)
                                              ↑ 现在可以触发 [06:00:00, 06:00:01) 窗口

Watermark 是一个特殊的时间戳：

当 Watermark W 经过算子时，表示「所有时间戳 < W 的事件都已到达」
事件时间窗口在 Watermark 越过窗口的 maxTimestamp（结束时间减 1ms）时触发计算
迟到的数据（时间戳 < Watermark 但后到达）如果所属窗口已关闭，会被丢弃或发到侧输出

Watermark 生成策略

有界乱序（Bounded Out-of-Orderness）

DataStream<Event> stream = env.fromSource(source,
    WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
        .withIdleness(Duration.ofMinutes(5))
        .withTimestampAssigner((event, timestamp) -> event.getEventTime()),
    "Kafka Source"
);

参数	含义
`Duration.ofSeconds(10)`	最大乱序时间：Watermark = 最大事件时间 - 10 秒 - 1ms（保证严格小于）
`withIdleness`	如果某个分区 5 分钟无数据，标记为 idle，不阻塞 Watermark
`withTimestampAssigner`	从事件中提取 Event Time 时间戳

单调递增（Monotonous Timestamps）

WatermarkStrategy.<Event>forMonotonousTimestamps()
    .withTimestampAssigner((event, ts) -> event.getEventTime());

假设数据严格按 Event Time 顺序到达（无乱序）。Watermark = 当前最大事件时间 - 1ms。

无 Watermark（Processing Time）

WatermarkStrategy.noWatermarks()

用于 Processing Time 场景。

自定义 Watermark 生成

WatermarkStrategy<Event> customStrategy = new WatermarkStrategy<Event>() {
    @Override
    public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context ctx) {
        return new WatermarkGenerator<Event>() {
            private long maxTimestamp = Long.MIN_VALUE;

            @Override
            public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
                maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
            }

            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                // 还没收到任何事件时必须跳过，否则会发出 Long.MIN_VALUE - 10000
                // 这种越界的 Watermark，下游窗口会立刻被判定为过期
                if (maxTimestamp == Long.MIN_VALUE) {
                    return;
                }
                // 每 200ms 发射一次 Watermark，落后最大事件时间 10 秒
                output.emitWatermark(new Watermark(maxTimestamp - 10000));
            }
        };
    }
};

onPeriodicEmit 默认每 200ms 调用一次，通过 env.getConfig().setAutoWatermarkInterval() 调整。

AllowedLateness

窗口触发后，仍允许迟到数据进入：

stream.keyBy(Event::getUserId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .allowedLateness(Duration.ofSeconds(30))
    .sideOutputLateData(lateOutputTag)
    .aggregate(new CountAggregate());

阶段	Watermark 位置	行为
窗口触发前	< 窗口结束时间	正常累积
窗口触发后	窗口结束时间 ~ 窗口结束 + 30s	迟到数据更新结果（增量输出）
窗口关闭后	> 窗口结束 + 30s	迟到数据丢弃或发送到 sideOutput

SideOutput（侧输出）

OutputTag<Event> lateTag = new OutputTag<Event>("late-data"){};

SingleOutputStreamOperator<Result> mainResult = stream
    .keyBy(Event::getUserId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .allowedLateness(Duration.ofSeconds(30))
    .sideOutputLateData(lateTag)
    .aggregate(new CountAggregate());

// 获取迟到数据
DataStream<Event> lateData = mainResult.getSideOutput(lateTag);
lateData.print("LATE");

空闲检测

如果某个 Kafka 分区长时间无数据，Watermark 不会推进，导致窗口永远不触发。

WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withIdleness(Duration.ofMinutes(5));  // 5 分钟无数据 = idle

idle 的分区不再参与 Watermark 计算，其他分区的 Watermark 可以正常推进。

时间语义对比

语义	Watermark	乱序容忍	可复现	延迟
Event Time	有	✅	✅	取决于 Watermark 延迟
Ingestion Time	有（单调递增）	❌	✅	低
Processing Time	无	❌	❌	最低

eventd v3 配置

eventd v3 使用 Event Time + 10 秒乱序容忍：

WatermarkStrategy<Event> strategy = WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withIdleness(Duration.ofMinutes(5))
    .withTimestampAssigner((event, ts) -> event.getEventTime());

DataStream<Event> stream = env.fromSource(kafkaSource, strategy, "events");

stream.keyBy(Event::getSymbol)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .allowedLateness(Duration.ofSeconds(10))
    .aggregate(new OrderCountAggregate());

10 秒乱序容忍：覆盖大多数网络延迟
1 分钟窗口：实时面板刷新粒度
10 秒 allowedLateness：迟到数据仍可计入

小结

概念	说明
Watermark	「不会再有更早数据」的时间标记
`forBoundedOutOfOrderness`	有界乱序，延迟 = 最大乱序时间
`forMonotonousTimestamps`	单调递增，无乱序
`withIdleness`	空闲分区检测
`allowedLateness`	窗口触发后仍接收迟到数据
`sideOutputLateData`	超期迟到数据的侧输出

下一节 4.3 Flink 窗口：滚动、滑动、会话、全局。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页