Flink 理解流式计算中的窗口概念

一、描述

Window 是处理无限流的核心。Flink 认为 Batch 是 Streaming 的一个特例，所以 Flink 底层的引擎是一个流式引擎，在上面实现了流处理和批处理。
而窗口（Window）就是从Streaming 到 batch 的一个桥梁。Flink 提供了非常完善的窗口机制，这是 Flink 最大的亮点之一（其他的亮点包括消息乱序处理和 Checkpoint 机制）

二、窗口的生命周期

窗口的生命周期，就是创建和销毁。
窗口的开始时间和结束时间是基于自然时间创建的，比如指定一个5s的窗口，那么1分钟内就会创建12个窗口。
什么时候窗口会被创建？当第一个元素进入到窗口开始时间的时候，这个窗口就被创建了。
什么时候窗口会被销毁？当时间（ProcessTime、EventTime或者 IngestionTime）越过了窗口的结束时间，再加上用户自定义的窗口延迟时间（allowed lateness），窗口就会被销毁。
举个例子来说，假设我们定义了一个基于事件时间的窗口，长度是5分钟，并且允许有1分钟的延迟。
当第一个元素包含了一个12:00的事件时间进来时，Flink会创建一个12:00 到 12:05 的窗口；在水位到 12:06 的时候，会销毁这个窗口。

每个窗口都会绑定一个触发器和一个执行函数。触发器定义了何时会触发窗口的执行函数的计算
，比如在窗口元素数量大于等于4的时候，或者水位经过了窗口结束时间的时候。
另外，每个窗口可以指定驱逐器（Evictor），它的作用是在触发器触发后，执行函数执行前，移除一些元素。

三、Keyed 和 Non-keyed Window

在定义窗口之前，首先要指定你的流是否应该被分区，使用 keyBy(…) 后，相同的 key 会被划分到不同的流里面，每个流可以被一个单独的 task 处理。如果不使用 keyBy ，所有数据会被划分到一个窗口里，只有一个task处理，并行度是1。

四、窗口的分类和选择

在指定了数据流是否分区之后，下一步是要去指定窗口的类型。窗口分配器（window assigner）定义了元素如何划分到不同的窗口中。
对于 keyed Streams，使用 window （…）来定义，对于非 keyed Streams，使用 windowAll（…）来定义。
Flink 预定义了很多种窗口类型，可以满足大多数日常使用需求：tumbling windows（翻滚窗口）, sliding windows（滑动窗口）, session windows（会话窗口） and global windows（全局窗口）。
所有内置的窗口（除了全局窗口）都是基于时间（ProcessTime或 EventTime）的。

1、Tumbling Windows

翻滚窗口有一个固定的长度，并且不会重复。比如，下图是指定了一个5分钟的翻滚窗口的样子：

（每个窗口都不重叠，每5分钟一个窗口）

// 例子1：tumbling event-time windows
// 定义一个数据流
val input: DataStream[T] = ...
// 这里的 key selector，如果是元组的化，可以使用_._1，如果是case class 可以使用字段名来指定
input.keyBy(<key selector>)
// 指定了一个TumblingEventTimeWindows，窗口大小为5分钟.window(TumblingEventTimeWindows.of(Time.seconds(5)))
// 窗口的操作.<windowed transformation>(<window function>)// 例子2：tumbling processing-time windows
input.keyBy(<key selector>).window(TumblingProcessingTimeWindows.of(Time.seconds(5))).<windowed transformation>(<window function>)// 例子3：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>)

在例子3中，TumblingEventTimeWindows.of 指定了第二个参数 offset，它的作用是改变窗口的时间。
如果我们指定了一个15分钟的窗口，那么每个小时内，每个窗口的开始时间和结束时间为：
[00:00,00:15)
[00:15,00:30)
[00:30,00:45)
[00:45,01:00)
如果我们指定了一个5分钟的offset，那么每个窗口的开始时间和结束时间为：
[00:05,00:20)
[00:20,00:35)
[00:35,00:50)
[00:50,01:05)
一个实际的应用场景是，我们可以使用 offset 使我们的时区以0时区为准。比如我们生活在中国，时区是
UTC+08:00，可以指定一个 Time.hour(-8)，使时间以0时区为准。

2、Slidding Windows

滑动窗口指定了两个参数，第一个参数是窗口大小，第二个参数控制了新的窗口开始的频率。
如果滑动距离小于窗口距离的话，那么一个元素可能被分配到多个窗口中。
比如，窗口大小10分钟，每5分钟滑动一次，如下图：

val input: DataStream[T] = ...// 例子1：sliding event-time windows
input.keyBy(<key selector>).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))).<windowed transformation>(<window function>)// 例子2：sliding processing-time windows
input.keyBy(<key selector>).window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))).<windowed transformation>(<window function>)// 例子3，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>)

例子3中，同样指定了一个 offset 参数，用来控制窗口开始的时间。

3、Session Windows

会话窗口根据会话的间隔来把数据分配到不同的窗口。
会话窗口不重叠，没有固定的开始时间和结束时间。

比如音乐 app 听歌的场景，我们想统计一个用户在一个独立的 session 中听了多久的歌曲（如果超过15分钟没听歌，那么就是一个新的 session 了）
我们可以用 spark Streaming ，每一个小时进行一次批处理，计算用户session的数据分布，但是 spark Streaming 没有内置对 session 的支持，我们只能手工写代码来维护每个 user 的 session 状态，里面仍然会有诸多的问题。
下一次会单独写一篇文章来讨论，如何使用flink 的 session window 来实现这个问题

4、Global Windows

全局 window 把所有相同 key 的数据，放到一个 window 来，它没有自然的窗口结束时间，所以我们需要自己指定触发器

val input: DataStream[T] = ...input.keyBy(<key selector>).window(GlobalWindows.create()).<windowed transformation>(<window function>)

五、写在最后

谢谢大家的阅读，原创不易，喜欢就随手点个赞，这将是对我最大的鼓励。如果你觉得文章还不错，就关注一下我的公众号：黑科技部落。获取Flink技术文档和万款软件。