Flink 彻底理解 window（窗口）

一、描述

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、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>)