你不得不知的Flink窗口

左手的ㄟ右手 2023-03-13 06:55 27阅读 0赞

# 一、Windows详解 #

如果你真的头铁看了上一篇的算子二，那么没有问题，看完这篇你应该会更加清晰的理解窗口，如果没看也没问题，我会适当的引入部分内容，但看完这篇后还是请打开算子二的窗口章节再去细品下相关算子！

## 1.1 图解关系 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center]

## 1.2 窗口操作类型分类 ##

对于窗口的操作分为两种，一种是keyedstrem，另一种是DataStream；他们的主要区别也仅仅在于建立窗口的时候一个为.window(…)，一个为.windowAll(…)。经过keyBy的数据流将形成多组数据，下游算子的多个实例可以并行计算。而对于windowAll不对数据流进行分组，所有数据将发送到下游算子单个实例上。

### 1.2.1 窗口的大致骨架 ###

// Keyed Window
    stream
           .keyBy(...)               <-  按照一个Key进行分组
           .window(...)              <-  将数据流中的元素分配到相应的窗口中
          [.trigger(...)]            <-  指定触发器Trigger（可选）
          [.evictor(...)]            <-  指定清除器Evictor(可选)
           .reduce/aggregate/process()      <-  窗口处理函数Window Function
    
    // Non-Keyed Window
    stream
           .windowAll(...)           <-  不分组，将数据流中的所有元素分配到相应的窗口中
          [.trigger(...)]            <-  指定触发器Trigger（可选）
          [.evictor(...)]            <-  指定清除器Evictor(可选)
           .reduce/aggregate/process()      <-  窗口处理函数Window Function

### 1.2.2 窗口分类 ###

窗口主要有两种，一种基于时间（Time-based Window），一种基于数量（Count-based Window）。这里主要讨论的事基于时间的窗口TimeWindow。每个TimeWindow都有一个开始时间和结束时间，表示一个左闭右开的时间段。而时间窗口有可细分为滚动窗口、滑动窗口和会话窗口，而每个窗口根据时间有可以分为事件时间（Event Time）、摄取事件（Ingestion Time）、处理时间（Processing Time）。（基于数量的窗口根据事件到达窗口的先后顺序管理窗口，到达窗口的先后顺序和Event Time并不一致，因此Count-based Window的结果具有不确定性）

#### 1.2.2.1 滚动窗口 ####

滚动窗口下窗口之间不重叠，且窗口长度是固定的。我们可以用TumblingEventTimeWindows和TumblingProcessingTimeWindows创建一个基于Event Time或Processing Time的滚动时间窗口。窗口的长度可以用org.apache.flink.streaming.api.windowing.time.Time中的seconds、minutes、hours和days来设置。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center 1]  
举例：

val input: DataStream[T] = ...
    
    // 事件时间滚动窗口5秒
    input
        .keyBy(...)
        .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .<window function>(...)
    
    // 处理时间滚动窗口5秒
    input
        .keyBy(...)
        .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .<window function>(...)
    
    // 事件时间滚动窗口1小时，时间偏移量30分钟
    input
        .keyBy(...)
        .window(TumblingEventTimeWindows.of(Time.hours(1), Time.minutes(30)))
        .<window function>(...)

注：有些代码可能使用的是timeWindow而非window timeWindow是一种简写。当我们在执行环境设置了TimeCharacteristic.EventTime时，Flink对应调用TumblingEventTimeWindows；如果我们基于TimeCharacteristic.ProcessingTime，Flink使用TumblingProcessingTimeWindows。

val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 从调用时刻开始给env创建的每一个stream追加时间特征
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val input: DataStream[T] = ...
    input.keyBy(...).timeWindow(Time.seconds(1))

#### 1.2.2.2 滑动窗口 ####

滑动窗口以一个步长（Slide）不断向前滑动，窗口的长度固定。使用时，我们要设置Slide和Size。Slide的大小决定了Flink以多大的频率来创建新的窗口，Slide较小，窗口的个数会很多。Slide小于窗口的Size时，相邻窗口会重叠，一个事件会被分配到多个窗口；Slide大于Size，有些事件可能被丢掉。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center 2]  
例：窗口时间间隔可以使用Time.milliseconds(x), Time.seconds(x), Time.minutes(x)中的一种定义，跟前面介绍的一样也可以用timeWindow代替

val input: DataStream[T] = ...
    // sliding event-time windows
    input
        .keyBy(...)
        .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
        .<window function>(...)
    // sliding processing-time windows
    input
        .keyBy(<...>)
        .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
        .<window function>(...)
    // sliding processing-time windows offset by -8 hours
    input
        .keyBy(<...>)
        .window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8)))
        .<window function>(...)

#### 1.2.2.3 会话窗口 ####

会话窗口根据Session gap切分不同的窗口，当一个窗口在大于Session gap的时间内没有接收到新数据时，窗口将关闭。在这种模式下，窗口的长度是可变的，每个窗口的开始和结束时间并不是确定的。我们可以设置定长的Session gap，也可以使用SessionWindowTimeGapExtractor动态地确定Session gap的长度。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center 3]  
例：

val input: DataStream[T] = ...
    // event-time session windows with static gap
    input
        .keyBy(...)
        .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
        .<window function>(...)
    // event-time session windows with dynamic gap
    input
        .keyBy(...)
        .window(EventTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor[T] {
          override def extract(element: T): Long = {
            // determine and return session gap
          }
        }))
        .<window function>(...)
    // processing-time session windows with static gap
    input
        .keyBy(...)
        .window(ProcessingTimeSessionWindows.withGap(Time.minutes(10)))
        .<window function>(...)
    // processing-time session windows with dynamic gap
    input
        .keyBy(...)
        .window(DynamicProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor[T] {
          override def extract(element: T): Long = {
            // determine and return session gap
          }
        }))
        .<window function>(...)

*  静态会话间隔可以使用Time.milliseconds(x), Time.seconds(x), Time.minutes(x)其中一种定义；
 *  动态会话间隔可以通过实现 SessionWindowTimeGapExtractor 接口；
 *  注意：由于session window会话窗口没有固定的开始和结束时间，因此它们的计算加工方式与

tumbling window和sliding window不同。 在内部，session window 算子为每个到达的记录创建一个新窗口，如果它们彼此之间的距离比定义的间隙更接近，则将窗口合并在一起。 为了可合并，session window 算子需要合并触发器和合并窗口函数，例如ReduceFunction，AggregateFunction或ProcessWindowFunction（FoldFunction无法合并。）

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center]: /images/20230312/c99c160655a3410d8eee578246824ee3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20200512173445179.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/2020051217350425.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20200512173518400.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NjU3OTA5,size_16,color_FFFFFF,t_70#pic_center