Flink滑动窗口增强

本节主要介绍Flink滑动窗口以及滑动窗口的优化方式。

Flink窗口的详细内容请参见官网：https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/operators/windows.html。

窗口介绍

窗口中数据的保存形式主要有中间结果和原始数据两种，对窗口中的数据使用公共算子，如sum等操作时（window(SlidingEventTimeWindows.of(Time.seconds(20), Time.seconds(5))).sum）仅会保留中间结果；当用户使用自定义窗口时（window(SlidingEventTimeWindows.of(Time.seconds(20), Time.seconds(5))).apply(new UDF)）保存所有的原始数据。

用户使用自定义SlidingEventTimeWindow和SlidingProcessingTimeWindow时，数据以多备份的形式保存。假设窗口的定义如下：

window(SlidingEventTimeWindows.of(Time.seconds(20), Time.seconds(5))).apply(new UDFWindowFunction)

当一个数据到来时，会被分配到20/5=4个不同的窗口中，即数据在内存中保存了4份。当窗口大小/滑动周期非常大时，冗余现象非常严重。

图1 窗口原始结构示例

假设一个数据在102秒时到来，它将会被分配到[85, 105)、[90, 110)、[95, 115)以及[100, 120)四个不同的窗口中。

窗口优化

针对上述SlidingEventTimeWindow和SlidingProcessingTimeWindow在保存原始数据时存在的数据冗余问题，对保存原始数据的窗口进行重构，优化存储，使其存储空间大大降低，具体思路如下：

1. 以滑动周期为单位，将窗口划分为若干相互不重合的pane。

   每个窗口由一到多个pane组成，多个pane对窗口构成了覆盖关系。所谓一个pane即一个滑动周期，如：在窗口window(SlidingEventTimeWindows.of(Time.seconds(20), Time.seconds.of(5)))中pane的大小为5秒，假设这个窗口为[100, 120)，则包含的pane为[100, 105), [105, 110), [110, 115), [115, 120)。

   图2 窗口重构示例
   
2. 当某个数据到来时，并不分配到具体的窗口中，而是根据自己的时间戳计算出该数据所属的pane，并将其保存到对应的pane中。

   一个数据仅保存在一个pane中，内存中只有一份。

   图3 窗口保存数据示例
   
3. 当需要触发某个窗口时，计算该窗口包含的所有pane，并取出合并成一个完整的窗口计算。
   图4 窗口触发计算示例
   
4. 当某个pane不再需要时，将其从内存中删除。
   图5 窗口删除示例
   

通过优化，可以大幅度降低数据在内存以及快照中的数量。