Flink中事件时间和处理时间(TumblingEventTimeWindows TumblingProcessingTimeWindows)

TumblingEventTimeWindows和TumblingProcessingTimeWindows是 Flink 中两种不同的窗口类型.

区别如下:

1. 时间类型：TumblingEventTimeWindows是基于事件时间的窗口类型，可以通过设置Watermark和EventTimeCharacteristic来确定事件时间；而TumblingProcessingTimeWindows是基于处理时间的窗口类型，时间由 Flink 运行时系统确定。

2. 窗口大小的选取：在TumblingEventTimeWindows中，窗口大小通常由用户设定的时间长度、Watermark和窗口策略共同决定；而在TumblingProcessingTimeWindows中，窗口大小通常由用户设定的时间长度和窗口策略共同决定。

3. 触发器的行为：在TumblingEventTimeWindows中，Watermark的到达会触发窗口计算；而在TumblingProcessingTimeWindows中，窗口计算会根据设定的 Trigger 来触发。

二者通过assignWindows方法就可以看出区别
TumblingEventTimeWindows的assignWindows源码内容如下

@Override
    public Collection<TimeWindow> assignWindows(
            Object element, long timestamp, WindowAssignerContext context) {
        if (timestamp > Long.MIN_VALUE) { // 区别1 先对事件时间时间进行判断
            if (staggerOffset == null) {
                staggerOffset =
                        windowStagger.getStaggerOffset(context.getCurrentProcessingTime(), size);
            }
            // Long.MIN_VALUE is currently assigned when no timestamp is present
            long start =
                    TimeWindow.getWindowStartWithOffset(
                            timestamp, (globalOffset + staggerOffset) % size, size);// 区别2 以事件时间作为参数传入
            return Collections.singletonList(new TimeWindow(start, start + size));
        } else {
            throw new RuntimeException(
                    "Record has Long.MIN_VALUE timestamp (= no timestamp marker). "
                            + "Is the time characteristic set to 'ProcessingTime', or did you forget to call "
                            + "'DataStream.assignTimestampsAndWatermarks(...)'?");
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

可以看到参数timestamp在TimeWindow.getWindowStartWithOffset(timestamp, (globalOffset + staggerOffset) % size, size)有使用到,timestamp就是事件时间.

TumblingProcessingTimeWindows的assignWindows源码内容如下

@Override
   public Collection<TimeWindow> assignWindows(
           Object element, long timestamp, WindowAssignerContext context) {
       final long now = context.getCurrentProcessingTime();
       // 区别1 这里没有对事件时间进行判断,因为不需要使用到事件时间
       if (staggerOffset == null) {
           staggerOffset =
                   windowStagger.getStaggerOffset(context.getCurrentProcessingTime(), size);
       }
       long start =
               TimeWindow.getWindowStartWithOffset(
                       now, (globalOffset + staggerOffset) % size, size);
       return Collections.singletonList(new TimeWindow(start, start + size)); // 区别2 这里以当前时间作为参数传入
   }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

从这段代码可以看到timestamp这个参数并没有进行使用在TimeWindow.getWindowStartWithOffset( now, (globalOffset + staggerOffset) % size, size);这段代码中第一个参数换成了now使用的当前时间,也就是处理时间.

使用场景:
TumblingEventTimeWindows适用于基于事件时间的数据流，并通常用于执行基于时间的聚合操作。例如，可以使用TumblingEventTimeWindows来计算每小时的平均输入请求量或处理不同餐厅菜品的每月营业额。

TumblingProcessingTimeWindows则适用于基于处理时间的数据流，并通常用于执行其他类型的基于时间的操作。例如，可以使用 TumblingProcessingTimeWindows来检测某个服务在过去一分钟内是否超过了阈值，或者使用计时器来触发某个事件。

根据数据流的特定需求和数据特性，选择适当的时间窗口类型非常重要。如果处理的是基于事件时间的数据流，并且需要按时间窗口对数据进行聚合，则应该选择 TumblingEventTimeWindows；如果处理的是基于处理时间的数据流，并且需要按时间进行其它类型的操作，则适合使用TumblingProcessingTimeWindows。