📚 课程目录第4章 Watermark与窗口的关系

4. Watermark与窗口的关系：滚动窗口、滑动窗口、会话窗口中的Watermark行为，窗口触发条件

窗口和Watermark，这两个东西在Flink里就像一对连体婴。你单独讲窗口，不讲Watermark，那窗口就是个摆设；你单独讲Watermark，不讲窗口，那Watermark就是个没用的时间戳。我刚开始学Flink那会儿，就犯过这个错——把窗口和Watermark分开理解，结果写出来的作业总是不按预期触发。

说白了，Watermark就是窗口的“发令枪”。枪不响，窗口就不动。今天我们就把这三种窗口——滚动、滑动、会话——和Watermark的配合关系彻底讲清楚。

  核心结论先放这： 窗口的触发时机 = max(窗口结束时间, 当前Watermark)。Watermark一旦越过窗口的结束时间，这个窗口就会被触发计算。

4.1 滚动窗口（Tumbling Window）

滚动窗口是最简单的。窗口之间不重叠，每个数据只属于一个窗口。比如你设置一个5分钟的滚动窗口，那[00:00, 00:05)、[00:05, 00:10)……就这么一路切下去。

那Watermark在这里扮演什么角色？

举个例子。你有一个5分钟的滚动窗口，结束时间是00:05:00。正常情况下，00:05:00一到，窗口就该触发了。但问题是，数据可能迟到。如果00:05:00到了，但00:04:55的数据还没来，你直接触发窗口，那这个数据就丢了。

Watermark就是来解决这个问题的。它告诉你：“现在我可以确定，所有时间戳早于00:05:00的数据都已经到了。” 当Watermark >= 00:05:00时，窗口才触发。

我个人的习惯： 滚动窗口的Watermark延迟设置，一般取业务容忍的最大延迟时间。比如业务说“晚3分钟的数据我还能接受”，那就设3秒的延迟？不对，是3分钟。别搞混了，单位要看清楚。

// 滚动窗口 + Watermark 示例
DataStream<SensorReading> stream = env
    .addSource(new SensorSource())
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<SensorReading>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
    );

stream
    .keyBy(r -> r.getId())
    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .apply(new MyWindowFunction())
    .print();

这段代码里，我设了5秒的乱序容忍度。也就是说，Watermark会比当前最大时间戳慢5秒。窗口结束时间到了，但Watermark没到，窗口就等着。等Watermark追上来，窗口才触发。

我曾经踩过的坑： 滚动窗口的结束时间是以整点对齐的。比如10秒窗口，从0秒开始，那窗口就是[0,10)、[10,20)……不是从第一条数据的时间开始算。这个和滑动窗口不一样，别搞混了。

4.2 滑动窗口（Sliding Window）

滑动窗口比滚动窗口复杂一点。它有窗口大小和滑动步长两个参数。比如窗口大小10分钟，滑动步长5分钟，那窗口就是[00:00, 00:10)、[00:05, 00:15)、[00:10, 00:20)……窗口之间是重叠的。

Watermark在滑动窗口里的行为，和滚动窗口本质上是一样的——Watermark越过窗口结束时间，窗口触发。但有一个细节要注意：一个数据可能属于多个窗口。

你想想看，如果一条数据的时间戳是00:07:00，它既属于[00:00, 00:10)这个窗口，也属于[00:05, 00:15)这个窗口。那这条数据会被复制到两个窗口里分别计算。Watermark推进的时候，这两个窗口的触发时间是不同的。

窗口类型	窗口大小	滑动步长	数据归属	触发条件
滚动窗口	10s	10s（等于窗口大小）	每个数据只属于1个窗口	Watermark >= 窗口结束时间
滑动窗口	10s	5s	每个数据可能属于多个窗口	Watermark >= 每个窗口各自的结束时间

我记得有一次做实时大屏，用的就是滑动窗口。窗口大小1分钟，滑动步长10秒。结果发现窗口触发频率特别高，每10秒就要触发6个窗口（因为1分钟窗口有6个重叠）。Watermark推进一次，可能同时触发好几个窗口。这个对性能有影响，要注意。

// 滑动窗口示例
stream
    .keyBy(r -> r.getId())
    .window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10)))
    .apply(new MyWindowFunction())
    .print();

避坑指南： 滑动窗口的窗口数量 = 窗口大小 / 滑动步长。如果这个值太大（比如窗口1小时，步长1秒），那同时存在的窗口就有3600个。每个数据要复制到3600个窗口里，内存压力巨大。我曾经见过一个案例，就是因为这个参数设置不合理，导致JobManager OOM。

4.3 会话窗口（Session Window）

会话窗口是最特殊的一种。它没有固定的窗口大小，而是根据数据的活跃程度来划分窗口。你设一个会话间隔（session gap），如果两条数据的时间差超过这个间隔，它们就属于不同的会话窗口。

那Watermark在会话窗口里怎么工作？

嗯，这里要注意。会话窗口的触发条件和其他窗口不太一样。它不是等Watermark越过某个固定的结束时间，而是等Watermark越过“当前会话窗口的结束时间 + 会话间隔”。

为什么？因为会话窗口的结束时间是不确定的。比如你有一个会话窗口，最后一条数据的时间是00:10:00，会话间隔是5分钟。那这个窗口的结束时间应该是00:15:00。但如果00:12:00又来了一条数据，那窗口的结束时间就延长到00:17:00了。

所以，会话窗口的触发条件是：Watermark >= 当前会话窗口的结束时间 + 会话间隔。这样能确保不会再有新数据加入这个会话。

  关键点： 会话窗口的合并机制。当两个会话窗口因为新数据的到来而“连在一起”时，Flink会把它们合并成一个更大的会话窗口。这个合并操作是在Watermark推进过程中触发的。

// 会话窗口示例
stream
    .keyBy(r -> r.getId())
    .window(EventTimeSessionWindows.withGap(Time.minutes(5)))
    .apply(new MyWindowFunction())
    .print();

我个人觉得，会话窗口是最难调优的。会话间隔设得太小，窗口切得太碎；设得太大，窗口又迟迟不触发。而且会话窗口的合并逻辑，在数据量大的时候，对状态后端压力很大。

4.4 三种窗口的触发条件对比

我把三种窗口的触发条件整理成了一张图，方便你对比理解。

4.5 窗口触发的完整流程

最后，我把窗口触发的完整流程总结一下。不管哪种窗口，触发逻辑都遵循这个步骤：

数据进入：数据到达后，根据时间戳分配到对应的窗口（可能多个）。
Watermark更新：每来一条数据，Watermark都可能推进。Flink会取当前所有分区的Watermark最小值作为全局Watermark。
窗口检查：每次Watermark更新后，Flink会检查所有未触发的窗口，看Watermark是否越过了窗口的结束时间。
窗口触发：如果Watermark >= 窗口结束时间（会话窗口还要加会话间隔），这个窗口就被触发，执行计算函数。
窗口清理：触发完成后，窗口的状态会被清理，释放内存。

注意： 窗口触发后，如果又有迟到数据进来（时间戳小于Watermark），默认情况下这些数据会被丢弃。如果你不想丢，可以用allowedLateness机制。这个我们后面章节会详细讲。

嗯，三种窗口和Watermark的关系，说白了就是一句话：Watermark是窗口的触发器，窗口是Watermark的消费者。理解了这个，你就能驾驭Flink的实时计算了。