10.1 window(窗口)

10.1.1 概述

10.1.2 Window类型

• 时间窗口（Time Window）
• 滚动时间窗口
• 滑动时间窗口
• 会话窗口

• 计数窗口（Count Window）
• 滚动计数窗口
• 滑动计数窗口

滚动窗口(Tumbling Windows)

• 依据固定的窗口长度对数据进行切分

• 时间对齐，窗口长度固定，没有重叠

滑动窗口(Sliding Windows)

• 可以按照固定的长度向后滑动固定的距离

• 滑动窗口由固定的窗口长度和滑动间隔组成

• 可以有重叠(是否重叠和滑动距离有关系)

• 滑动窗口是固定窗口的更广义的一种形式，滚动窗口可以看做是滑动窗口的一种特殊情况（即窗口大小和滑动间隔相等）

会话窗口(Session Windows)

• 由一系列事件组合一个指定时间长度的timeout间隙组成，也就是一段时间没有接收到新数据就会生成新的窗口

• 特点：时间无对齐

10.2 Window API

10.2.1 概述

• 窗口分配器——window()方法

• 我们可以用.window()来定义一个窗口，然后基于这个window去做一些聚合或者其他处理操作。
注意window()方法必须在keyBy之后才能使用。

• Flink提供了更加简单的.timeWindow()和.countWindow()方法，用于定义时间窗口和计数窗口。

窗口分配器(window assigner)

• window()方法接收的输入参数是一个WindowAssigner

• WindowAssigner负责将每条输入的数据分发到正确的window中

• Flink提供了通用的WindowAssigner
• 滚动窗口（tumbling window）
• 滑动窗口（sliding window）
• 会话窗口（session window）
• 全局窗口（global window）

创建不同类型的窗口

• 滚动时间窗口（tumbling time window）
.timeWindow(Time.seconds(15))
• 滑动时间窗口（sliding time window）
.timeWindow(Time.seconds(15),Time.seconds(5))
• 会话窗口（session window）
.window(EventTimeSessionWindows.withGap(Time.minutes(10)))
• 滚动计数窗口（tumbling count window）
.countWindow(5)
• 滑动计数窗口（sliding count window）
.countWindow(10,2)
DataStream的windowAll()类似分区的global操作，这个操作是non-parallel的(并行度强行为1)，所有的数据都会被传递到同一个算子operator上，官方建议如果非必要就不要用这个API

10.2.2 TimeWindow

TimeWindow将指定时间范围内的所有数据组成一个window，一次对一个window里面的所有数据进行计算。
 

滚动窗口

Flink默认的时间窗口根据ProcessingTime进行窗口的划分，将Flink获取到的数据根据进入Flink的时间划分到不同的窗口中。
时间间隔可以通过Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。

滑动窗口

滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是window_size，一个是sliding_size。
下面代码中的sliding_size设置为了5s，也就是说，每5s就计算输出结果一次，每一次计算的window范围是15s内的所有元素。
时间间隔可以通过Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。

10.2.3 CountWindow

CountWindow根据窗口中相同key元素的数量来触发执行，执行时只计算元素数量达到窗口大小的key对应的结果。
 注意：CountWindow的window_size指的是相同Key的元素的个数，不是输入的所有元素的总数。

滚动窗口

默认的CountWindow是一个滚动窗口，只需要指定窗口大小即可，当元素数量达到窗口大小时，就会触发窗口的执行。

滑动窗口

滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是window_size，一个是sliding_size。
下面代码中的sliding_size设置为了2，也就是说，每收到两个相同key的数据就计算一次，每一次计算的window范围是10个元素。



10.2.4 window function

window function 定义了要对窗口中收集的数据做的计算操作，主要可以分为两类：
• 增量聚合函数（incremental aggregation functions）
• 全窗口函数（full window functions）

增量聚合函数

• 每条数据到来就进行计算，保持一个简单的状态。（来一条处理一条，但是不输出，到窗口临界位置才输出）
• 典型的增量聚合函数有ReduceFunction, AggregateFunction。

全窗口函数

• 先把窗口所有数据收集起来，等到计算的时候会遍历所有数据。（来一个放一个，窗口临界位置才遍历且计算、输出）
• ProcessWindowFunction，WindowFunction。

10.2.5 其它可选API

• .trigger() ——触发器
定义window 什么时候关闭，触发计算并输出结果
• .evitor() ——移除器
定义移除某些数据的逻辑
• .allowedLateness() ——允许处理迟到的数据
• .sideOutputLateData() ——将迟到的数据放入侧输出流
• .getSideOutput() ——获取侧输出流



10.2.6 代码测试

scala测试

测试滚动时间窗口的增量聚合函数

增量聚合函数，特点即每次数据过来都处理，但是到了窗口临界才输出结果。
• 编写java代码
本地开启socket服务
nc -lk 7777
启动Flink程序，在socket窗口输入数据输入(下面用“换行”区分每个15s内的输入，实际输入时无换行)
输出（下面用“换行”区分每个15s内的输出，实际输出无换行）因为代码实现每15s一个window，所以"sensor_1"中间一组才累计2，最初一次不累计，最后一次也是另外的window，重新从1计数。

测试滚动时间窗口的全窗口函数

全窗口函数，特点即数据过来先不处理，等到窗口临界再遍历、计算、输出结果。
• 编写java测试代码
• 启动本地socket
• 在本地socket输入，查看Flink输出结果
• 输入（以“空行”表示每个15s时间窗口内的输入，实际没有“空行”）
输出（以“空行”表示每个15s时间窗口内的输入，实际没有“空行”）
这里每个window都是分开计算的，所以第一个window里的sensor_1和第二个window里的sensor_1并没有累计。

测试滑动计数窗口的增量聚合函数

滑动窗口，当窗口不足设置的大小时，会先按照步长输出。
eg：窗口大小10，步长2，那么前5次输出时，窗口内的元素个数分别是（2，4，6，8，10），再往后就是10个为一个窗口了。
• 编写java代码：
这里获取每个窗口里的温度平均值
• 启动socket服务
• 本地socket输入，Flink控制台查看输出结果
• 输入
这里为了方便，就只输入同一个keyBy组的数据sensor_1
• 输出
输入时，会发现，每次到达一个窗口步长（这里为2），就会计算得出一次结果。
第一次计算前2个数的平均值
第二次计算前4个数的平均值
第三次计算前6个数的平均值
第四次计算前8个数的平均值
第五次计算前10个数的平均值
第六次计算前最近10个数的平均值
第七次计算前最近10个数的平均值
 
// 3. 其他可选API