Flink窗口概述及用法详解

快来打我* 2023-03-03 06:29 26阅读 0赞

## 一.简介 ##

窗口window是处理无限流的核心。窗口将流分成有限大小的“存储块”，我们可以在其上应用计算。本博客重点介绍如何在Flink中执行窗口化，以及程序员如何从其提供的功能中获得最大收益。

窗口式Flink程序的一般结构如下所示。第一个片段指的是键控流，而第二个片段指的是非键控流。正如人们所看到的，唯一的区别是keyBy(…)使用window操作键控流和使用windowAll(…)操作的非键控流。  
键控窗口

stream
           .keyBy(...)               <-  keyed versus non-keyed windows
           .window(...)              <-  required: "assigner"
          [.trigger(...)]            <-  optional: "trigger" (else default trigger)
          [.evictor(...)]            <-  optional: "evictor" (else no evictor)
          [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
          [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
           .reduce/aggregate/fold/apply()      <-  required: "function"
          [.getSideOutput(...)]      <-  optional: "output tag"

非键控窗口

stream
           .windowAll(...)           <-  required: "assigner"
          [.trigger(...)]            <-  optional: "trigger" (else default trigger)
          [.evictor(...)]            <-  optional: "evictor" (else no evictor)
          [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
          [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
           .reduce/aggregate/fold/apply()      <-  required: "function"
          [.getSideOutput(...)]      <-  optional: "output tag"

在上面，方括号（\[…\]）中的命令是可选的。这表明Flink允许以多种不同的方式自定义窗口逻辑，从而使其最适合各种的需求。

## 二.窗口生命周期 ##

简而言之，一旦应属于该窗口的第一个元素到达，就会创建一个窗口，并且当时间【事件时间或处理时间】超过其结束时间戳加上用户指定的时间延迟后，该窗口将被完全删除。Flink保证只删除基于时间的窗口，而不保证其他类型【例如全局窗口】的删除。例如，采用基于事件时间的开窗策略，该策略每5分钟创建一次不重叠【或翻滚】的窗口，并允许延迟1分钟。allowed lateness12:00~12:05，当带有时间戳的第一个元素落入此间隔时Flink将创建一个新窗口，当水印通过12:06 时间戳时，它将删除它。

此外，每个窗口将具有Trigger【触发器】和一个函数【ProcessWindowFunction、ReduceFunction、AggregateFunction或FoldFunction】连接到它。该函数将包含要应用于窗口内容的计算，而该Trigger指定了在什么条件下可以将窗口视为要应用该函数的条件。触发策略可能类似于“当窗口中的元素数大于4时”或“当水印通过窗口末尾时”。触发器还可以决定在创建和删除窗口之间的任何时间清除窗口的内容。在这种情况下，清除仅是指窗口中的元素，而不是窗口元数据。这意味着仍可以将新数据添加到该窗口。

除上述内容外，还可以指定一个Evictor，它将能够在触发触发器之后以及应用此功能之前/之后从窗口中删除元素。

## 三.键控与非键控窗口 ##

要指定的第一件事是流是否是键控流。这必须在定义窗口之前完成。使用keyBy(…)会将无限流分割成逻辑键流。如果keyBy(…)未调用，则不会将流转换为键控流。

在使用键控流的情况下，传入事件的任何属性都可以用作键。拥有键控流将使窗口化计算可以由多个任务并行执行，因为每个逻辑键控流都可以独立于其余逻辑流进行处理。引用同一键的所有元素将被发送到同一并行任务。

对于非键控流，原始流将不会拆分为多个逻辑流，并且所有窗口逻辑将由单个任务执行，即并行度为1。

## 四.窗口分配器 ##

指定流是否为键控流后，下一步是定义窗口分配器。窗口分配器定义了如何将元素分配给窗口。这是通过WindowAssigner 在window(…)【针对键控流】或windowAll()【针对非键控流】调用中指定的选项来完成的。

WindowAssigner负责将每个传入元素分配给一个或多个窗口。Flink带有针对最常见用例的预定义窗口分配器，即滚动窗口， 滑动窗口，会话窗口和全局窗口。还可以通过扩展WindowAssigner类来实现自定义窗口分配器。所有内置窗口分配器【全局窗口除外】均基于时间将元素分配给窗口，时间可以是处理时间，也可以是事件时间。

基于时间的窗口具有开始时间戳【包括端点】和结束时间戳【包括端点】，它们共同描述了窗口的大小。在代码中，Flink在使用TimeWindow基于时间的窗口时使用，该方法具有查询开始和结束时间戳记的方法maxTimestamp()，还具有返回给定窗口允许的最大时间戳的附加方法。

**滚动窗户**  
滚动窗口分配器受根据设置的窗口大小让每个元素到指定的窗口。滚动窗口具有固定的大小，并且不重叠。例如，如果指定大小为5分钟的滚动窗口，则将评估当前窗口，并且每五分钟将启动一个新窗口，如下图所示：  
![在这里插入图片描述][20200804100443571.png]  
以下代码段显示了如何使用滚动窗口：

val input: DataStream[T] = ...
    
    // tumbling event-time windows
    input
        .keyBy(<key selector>)
        .window(TumblingEventTimeWindows.of(Time.seconds(5)))
        .<windowed transformation>(<window function>)
    
    // tumbling processing-time windows
    input
        .keyBy(<key selector>)
        .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
        .<windowed transformation>(<window function>)
    
    // daily tumbling event-time windows offset by -8 hours.
    input
        .keyBy(<key selector>)
        .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
        .<windowed transformation>(<window function>)

时间间隔可以通过使用一个指定Time.milliseconds(x)，Time.seconds(x)， Time.minutes(x)，等等。

滚动窗口分配器还采用一个可选offset 参数，该参数可用于更改窗口的对齐方式。例如，如果没有偏移，则每小时滚动窗口与epoch对齐，即将获得诸如的窗口 1:00:00.000 - 1:59:59.999，2:00:00.000 - 2:59:59.999依此类推。如果要更改，可以提供一个偏移量。随着15分钟的偏移量，你会，例如，拿 1:15:00.000 - 2:14:59.999，2:15:00.000 - 3:14:59.999等一个重要的用例的偏移是窗口调整到比UTC-0时区等。例如，在中国，必须指定的偏移量Time.hours(-8)。

**滑动窗口**  
滑动窗口分配器以固定长度的创建窗口。类似于滚动窗口分配器，窗口的大小由窗口大小参数配置。附加的窗口滑动参数控制滑动窗口启动的频率。因此，如果滑动间隔小于窗口大小，则滑动窗口可能会重叠。在这种情况下，元素被分配给多个窗口。

例如，可以将大小为10分钟的窗口滑动5分钟。这样，每隔5分钟就会得到一个窗口，其中包含最近10分钟内到达的事件，如下图所示：  
![在这里插入图片描述][20200804103411279.png]

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

时间间隔可以通过使用一个指定Time.milliseconds(x)，Time.seconds(x)， Time.minutes(x)，等等。

滑动窗口分配器还采用一个可选offset参数，该参数可用于更改窗口的对齐方式。例如，在没有偏移的情况下，每小时滑动30分钟的窗口将与epoch对齐，即您将获得诸如的窗口 1:00:00.000 - 1:59:59.999，1:30:00.000 - 2:29:59.999依此类推。如果要更改，可以提供一个偏移量。随着15分钟的偏移量，你会，例如，拿 1:15:00.000 - 2:14:59.999，1:45:00.000 - 2:44:59.999等一个重要的用例的偏移是窗口调整到比UTC-0时区等。例如，在中国，必须指定的偏移量Time.hours(-8)。

**会话窗口**  
会话窗口按活动的会话组中的元素创建窗口。与滚动窗口和滑动窗口相比，会话窗口不重叠且没有固定的开始和结束时间。相反，当会话窗口在一定时间段内未收到元素时（即，发生不活动间隙时），它将关闭。会话窗口分配器可与静态配置会话间隙或会话间隙提取功能，其限定不活动周期有多长。当该时间段到期时，当前会话将关闭，随后的元素将分配给新的会话窗口。  
![在这里插入图片描述][20200804104258533.png]  
代码如下：

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

静态间隙可以通过使用一个来指定Time.milliseconds(x)，Time.seconds(x)， Time.minutes(x)，等。

动态间隙是通过实现SessionWindowTimeGapExtractor接口指定的。

**注意**：由于会话窗口没有固定的开始和结束，因此对它们的评估方式不同于滚动窗口和滑动窗口。在内部，会话窗口运算符会为每个到达的记录创建一个新窗口，如果窗口彼此之间的距离比已定义的间隔小，则将它们合并在一起。为了可合并的，会话窗口操作需要一个合并触发器以及合并的窗口函数，如ReduceFunction、AggregateFunction或ProcessWindowFunction 【FoldFunction不能合并】。

[20200804100443571.png]: /images/20230208/09e82b62810140599d0e69166530661e.png
[20200804103411279.png]: /images/20230208/7c49f8135a9b4543827eb72ea0acf06c.png
[20200804104258533.png]: /images/20230208/eff882b37db54b0aafc78be2b94508b8.png