Flink 的窗口计算

1.概述

Flink是一种流式计算引擎，主要是来处理无界数据流的，数据源源不断、无穷无尽。想要更加方便高效地处理无界流，一种方式就是将无限数据切割成有限的范围进行处理，这就是窗口计算。

2. 窗口

2.1 窗口分类

2.1.1 按照驱动类型分类

时间窗口：以时间点来定义窗口的开始和结束，所以截取出的是就是某一时间段的数据。

计数窗口：基于元素的个数来截取数据，到达固定的个数时就触发计算并关闭窗口。

2.1.2 按照窗口分配数据的规则分类

滚动窗口（Tumbling Windows）：滚动窗口有固定的大小，是一种对数据进行“均匀切片”的划分方式。窗口之间没有重叠，也不会有间隔，是收尾相接的状态。

滑动窗口（Sliding Windows）：滑动窗口除了窗口大小（window size），还有一个滑动步长（window slide）。滑动步长代表了窗口计算的频率。

会话窗口（Session Windows）：如果相邻两个数据到来的时间间隔（Gap）小于指定的大小（size），保持会话，它们属于同一个窗口；如果 gap 大于 size，那么新来的数据就属于新的会话窗口。所以最重要的参数就是会话的超时时间，也就是两个会话窗口之间的最小距离。

全局窗口：会把相同 key 中的所有数据分配到同一个窗口。这种窗口没有结束的时候，默认是不会触发计算的。如果希望对数据进行处理，需要自定义触发器。

2.2 窗口分配器

窗口分配器最通用的定义方式，就是调用 .window()方法。这个方法需要传入一个 WindowAssigner 作为参数，返回 WindowedStream。如果是非按键分组窗口，那么直接调用 .windowAll()方法，同样传入一个 WindowAssigner，返回的是 AllWindowedStream。

2.2.1 时间窗口

（1）滚动处理/事件时间窗口

stream.keyBy(...).window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(5))).aggregate(...)

stream.keyBy(...).window(SlidingEventTimeWindows.of(Time.seconds(10)，Duration.ofSeconds(5))).aggregate(...)

（2）滑动处理/事件时间窗口

stream.keyBy(...).window(SlidingProcessingTimeWindows.of(Time.seconds(10)，Duration.ofSeconds(5))).aggregate(...)

stream.keyBy(...).window(SlidingEventTimeWindows.of(Time.seconds(10)，Duration.ofSeconds(5))).aggregate(...)

（3）处理/事件时间会话窗口

stream.keyBy(...).window(ProcessingTimeSessionWindows.withGap(Duration.ofSeconds(10))).aggregate(...)

stream.keyBy(...).window(EventTimeSessionWindows.withGap(Duration.ofSeconds(10))).aggregate(...)

另外，还可以调用withDynamicGap()方法定义session gap的动态提取逻辑。

2.2.2 计数窗口

（1）滚动计数窗口

stream.keyBy(...).countWindow(10)

（2）滑动计数窗口

stream.keyBy(...).countWindow(10，3)

（3）全局窗口

stream.keyBy(...).window(GlobalWindows.create())

需要注意使用全局窗口，必须自行定义触发器才能实现窗口计算，否则起不到任何作用。

2.3 窗口函数

2.4 其他 API

2.4.1 触发器（Trigger）

stream.keyBy(...).window(...).trigger(new MyTrigger())

2.4.2 移除器（Evictor）

stream.keyBy(...).window(...).evictor(new MyEvictor())

3. 时间语义

事件时间：数据产生的时间。从Flink1.12版本开始，Flink已经将事件时间作为默认的时间语义了。

处理时间：数据真正被处理的时间

4. 水位线

在Flink中，用来衡量事件时间进展的标记，就被称作“水位线”（Watermark）。

4.1 水位线的特性

• 水位线是插入到数据流中的一个标记，可以认为是一个特殊的数据
• 水位线主要的内容是一个时间戳，用来表示事件时间的进展
• 水位线是基于数据的时间戳生成的
• 水位线的时间戳必须是单调递增的，以确保任务的时间时钟一直向前推进
• 水位线可以通过设置延迟，来保证正确处理乱序数据
• 一个水位线 t，表示在当前流中的事件时间已经达到了时间戳 t，这代表 t 之前的所有数据都到齐了，之后流中不会出现时间戳小于 t 的数据

4.2 水位线和窗口的工作原理

在 Flink 中，窗口是一个由水位线划定的时间框，其中窗口的数据应该放在对应的窗口桶里面。时间框里面的数据不一定符合时间范围要求，但是窗口桶里面的数据是绝对符合时间范围要求的。

注意：Flink中窗口桶并不是静态准备好的，而是动态创建——当有落在这个窗口桶范围的数据达到时，创建对应的窗口桶。另外，水位线到达窗口桶的结束时间时，窗口桶就触发计算并关闭。

4.3 生成水位线

Flink中的水位线是流处理中对低延迟和结果正确性的一个权衡机制。

4.3.1 水位线生成策略

1）有序流中内置水位线策略设置--单调递增

// 1.定义Watermark策略
WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
        // 1.1 指定watermark生成：升序的watermark，没有等待时间
        .<WaterSensor>forMonotonousTimestamps()
        // 1.2 指定 时间戳分配器，从数据中提取
        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                // 返回的时间戳，要 毫秒
               ...
            }
        });

2）乱序流中内置水位线策略设置--有界乱序

// 1.定义Watermark策略
WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
        // 1.1 指定watermark生成：乱序的，等待3s
        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
        // 1.2 指定 时间戳分配器，从数据中提取
        .withTimestampAssigner((element, recordTimestamp) -> {
            // 返回的时间戳，要 毫秒
            ...
        });

3）指定 Watermark 策略

SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);

4.3.2 自定义水位线生成器

class MyGenerator implements WatermarkGenerator<WaterSensor>{

    long maxTx;
    long outTx;

    public MyGenerator(long outTx) {
        this.outTx = outTx;
    }

    @Override
    public void onEvent(WaterSensor ws, long eventTimestamp, WatermarkOutput output) {
        maxTx = Math.max(maxTx, ws.ts);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTx - outTx - 1));
    }
}

SingleOutputStreamOperator<WaterSensor> withWatermarkDS = wsDS.assignTimestampsAndWatermarks(
        WatermarkStrategy.forGenerator(
                new WatermarkGeneratorSupplier<WaterSensor>() {
                    @Override
                    public WatermarkGenerator<WaterSensor> createWatermarkGenerator(Context context) {
                        return new MyGenerator(5);
                    }
                }
        ).withTimestampAssigner(
                new SerializableTimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor ws, long recordTimestamp) {
                        return ws.getTs();
                    }
                }
        )
);

1）周期性水位线生成器（Periodic Generator）

周期性生成器一般是通过onEvent()观察判断输入的事件，而在onPeriodicEmit()里发出水位线。

2）断点式水位线生成器（Punctuated Generator）

断点式生成器会不停地检测onEvent()中的事件，当发现带有水位线信息的事件时，就立即发出水位线。所以把发射水位线的逻辑写在onEvent方法当中即可。

3）在数据源中发送水位线

DataStreamSource<String> fromSource = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");

4.4 水位线的传递

如果上游一个并行度，下游多个并行度，广播

如果上游多个并行度，下游一个并行度，将上游所有并行度的水位线取最小

如果上游多个并行度，下游多个并行度，先广播，再取最小

4.5 迟到数据的处理

4.5.1 推迟水位线

WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10));

4.5.2 设置窗口桶延时关闭

.window(TumblingEventTimeWindows.of(Time.seconds(5))).allowedLateness(Time.seconds(3))

4.5.3 使用侧流接收迟到的数据

OutputTag<WaterSensor> lateTag = new OutputTag<>("s1", Types.POJO(WaterSensor.class)){};

.window(TumblingEventTimeWindows.of(Time.seconds(5))).allowedLateness(Time.seconds(3)).sideOutputLateData(lateTag).getSideOutput(lateTag).printToErr("关窗后的迟到数据");

5. 双流联结

5.1 窗口联结

stream1.join(stream2).where(<KeySelector>).equalTo(<KeySelector>).window(<WindowAssigner>).apply(<JoinFunction>)

5.2 间隔联结

stream1.keyBy(<KeySelector>).intervalJoin(stream2.keyBy(<KeySelector>)).between(Time.milliseconds(-2), Time.milliseconds(1)).process (<ProcessJoinFunction>)