FLINK基础（105）: DS算子与窗口（16）窗口 (1) 简介

1 Windows定义

按固定时间区间计算该区间的值，例如15s计算汇总一次：

无穷的流，数据不间断的，例如有累计数据的需求，按上图的逻辑是处理不到的。换一种思路，每隔 15 秒，我们都将与上一次的结果进行 sum 操作（滑动聚合）：

流是无界的，我们不能限制流，所以上述方案也解决不了需求，但可以在有一个有界的范围内处理无界的流数据。

那么按一分钟一个时间窗口计算，相当于一个定义了一个 Window（窗口），window 的界限是1分钟，且每分钟内的数据互不干扰，因此也可以称为翻滚（不重合）窗口

第一分钟的数量为8，第二分钟是22，第三分钟是27。。。这样，1个小时内会有60个window。

再考虑一种情况，每30秒统计一次过去1分钟的数量之和：

通常来讲，Window 就是用来对一个无限的流设置一个有限的集合，在有界的数据集上进行操作的一种机制。window 又可以分为基于时间（Time-based）的 window 以及基于数量（Count-based）的 window。

2 Flink窗口类型

对于窗口的操作主要分为两种，分别对于Keyedstream和Datastream。他们的主要区别也仅仅在于建立窗口的时候一个为.window(...)，一个为.windowAll(...)。对于Keyedstream的窗口来说，他可以使得多任务并行计算，每一个logical key stream将会被独立的进行处理。

stream
       .keyBy(...)               <-  keyed versus non-keyed windows
       .window(...)/.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"

• 按照窗口的Assigner来分，窗口可以分为
  Tumbling window， sliding window，session window，global window，custom window

• 每种窗口又可分别基于processing time和event time。

• 还有一种window叫做count window，依据元素到达的数量进行分配，之后也会提到。

• 窗口的生命周期开始在第一个属于这个窗口的元素到达的时候，结束于第一个不属于这个窗口的元素到达的时候。

3 窗口的操作

3.1 Tumbling window

固定相同间隔分配窗口，每个窗口之间没有重叠。

 

WindowedStream<MovieRate, Integer, TimeWindow> Rates = rates
    .keyBy(MovieRate::getUserId)
    .window(TumblingEventTimeWindows.of(Time.milliseconds(3)));

3.2 Sliding Windows

跟上面一样，固定相同间隔分配窗口，只不过每个窗口之间有重叠。窗口重叠的部分如果比窗口小，窗口将会有多个重叠，即一个元素可能被分配到多个窗口里去。

WindowedStream<MovieRate, Integer, TimeWindow> Rates = rates
                .keyBy(MovieRate::getUserId)
                .window(SlidingEventTimeWindows.of(Time.milliseconds(10), Time.milliseconds(5)));

3.3 Session window

这种窗口主要是根据活动的事件进行窗口化，他们通常不重叠，也没有一个固定的开始和结束时间。一个session window关闭通常是由于一段时间没有收到元素。在这种用户交互事件流中，我们首先想到的是将事件聚合到会话窗口中（一段用户持续活跃的周期），由非活跃的间隙分隔开。

// 静态间隔时间
WindowedStream<MovieRate, Integer, TimeWindow> Rates = rates
                .keyBy(MovieRate::getUserId)
                .window(EventTimeSessionWindows.withGap(Time.milliseconds(10)));
// 动态时间
WindowedStream<MovieRate, Integer, TimeWindow> Rates = rates
                .keyBy(MovieRate::getUserId)
                .window(EventTimeSessionWindows.withDynamicGap(()));

3.4 Global window

将所有相同keyed的元素分配到一个窗口里。好吧，就这样：

WindowedStream<MovieRate, Integer, GlobalWindow> Rates = rates
    .keyBy(MovieRate::getUserId)
    .window(GlobalWindows.create());

4 Flink 的窗口机制

• 到达窗口操作符的元素被传递给 WindowAssigner。WindowAssigner 将元素分配给一个或多个窗口，可能会创建新的窗口。窗口本身只是元素列表的标识符，它可能提供一些可选的元信息，例如 TimeWindow 中的开始和结束时间。注意，元素可以被添加到多个窗口，这也意味着一个元素可以同时在多个窗口存在。

• 每个窗口都拥有一个 Trigger(触发器)，该 Trigger(触发器) 决定何时计算和清除窗口。当先前注册的计时器超时时，将为插入窗口的每个元素调用触发器。在每个事件上，触发器都可以决定触发(即、清除(删除窗口并丢弃其内容)，或者启动并清除窗口。一个窗口可以被求值多次，并且在被清除之前一直存在。注意，在清除窗口之前，窗口将一直消耗内存。

• 当 Trigger(触发器) 触发时，可以将窗口元素列表提供给可选的 Evictor，Evictor 可以遍历窗口元素列表，并可以决定从列表的开头删除首先进入窗口的一些元素。然后其余的元素被赋给一个计算函数，如果没有定义 Evictor，触发器直接将所有窗口元素交给计算函数。

• 计算函数接收 Evictor 过滤后的窗口元素，并计算窗口的一个或多个元素的结果。 DataStream API 接受不同类型的计算函数，包括预定义的聚合函数，如 sum（），min（），max（），以及 ReduceFunction，FoldFunction 或 WindowFunction。

  1. 窗口函数就是这四个：ReduceFunction，AggregateFunction，FoldFunction，ProcessWindowFunction。前两个执行得更有效，因为Flink可以增量地聚合每个到达窗口的元素。
  2. Flink必须在调用函数之前在内部缓冲窗口中的所有元素，所以使用ProcessWindowFunction进行操作效率不高。不过ProcessWindowFunction可以跟其他的窗口函数结合使用，其他函数接受增量信息，ProcessWindowFunction接受窗口的元数据。

这些是构成 Flink 窗口机制的组件。 接下来我们逐步演示如何使用 DataStream API 实现自定义窗口逻辑。 我们从 DataStream [IN] 类型的流开始，并使用 key 选择器函数对其分组，该函数将 key 相同类型的数据分组在一块。

SingleOutputStreamOperator<xxx> data = env.addSource(...);
data.keyBy();

5 Window的生命周期

5.1 创建：当属于该窗口的第一个元素到达时就会创建该窗口

5.2 销毁：当时间(event/process time)超过窗口的结束时间戳+用户指定的延迟时(allowedLateness(<time>))，窗口将被移除(仅限time-based window)

例如:对于一个每5分钟创建Tumbling Windows(即翻滚窗口)窗口，允许1分钟的时延，Flink将会在12:00到12:05这段时间内第一个元素到达时创建窗口，当watermark超过12:06时，该窗口将被移除。

5.3 Trigger(触发器)
指定了窗口函数在什么条件下可被触发，触发器还可以决定在创建和删除窗口之间的任何时间清除窗口的内容。在这种情况下，清除仅限于窗口中的元素，而不是窗口元数据（就是不删除窗口）。这意味着新数据仍然可以添加到该窗口中。

例如：当窗口中的元素个数超过4个时― 或者 ‖当水印达到窗口的边界时―触发计算

5.4 Window的函数
函数里定义了应用于窗口(Window)内容的计算逻辑

5.5 Evictor(驱逐者)
将在触发器触发之后或者在函数被应用前后，清除窗口中的元素

作者：木戎
链接：https://www.jianshu.com/p/ce1e009f8a95
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。