Window意为窗口。在流处理系统中数据源源不断流入到系统，我们可以逐条处理流入的数据，也可以按一定规则一次处理流中的多条数据。当处理数据时程序需要知道什么时候开始处理、处理哪些数据。窗口提供了这样一种依据，决定了数据何时开始处理。

Flink内置Window

Flink有3个内置Window

• 以事件数量驱动的Count Window

• 以会话间隔驱动的Session Window

• 以时间驱动的Time Window

本文围绕这3个内置窗口展开讨论，我们首先了解这3个窗口在运行时产生的现象，最后再讨论它们的实现原理。

Count Window

计数窗口，采用事件数量作为窗口处理依据。计数窗口分为滚动和滑动两类，使用keyedStream.countWindow实现计数窗口定义。

• Tumbling Count Window 滚动计数窗口
• 例子：以用户分组，当每位用户有3次付款事件时计算一次该用户付款总金额。下图中“消息A、B、C、D”代表4位不同用户，我们以A、B、C、D分组并计算金额。

/** 每3个事件，计算窗口内数据 */ keyedStream.countWindow(3);

• Sliding Count Window 滑动计数窗口
• 例子：一位用户每3次付款事件计算最近4次付款事件总金额。

/** 每3个事件，计算最近4个事件消息 */ keyedStream.countWindow(4,3);

Session Window

会话窗口，采用会话持续时长作为窗口处理依据。设置指定的会话持续时长时间，在这段时间中不再出现会话则认为超出会话时长。

例子：每只股票超过2秒没有交易事件时计算窗口内交易总金额。下图中“消息A、消息B”代表两只不同的股票。

/** 会话持续2秒。当超过2秒不再出现会话认为会话结束 */ keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(2)))

Time Window

时间窗口，采用时间作为窗口处理依据。时间窗分为滚动和滑动两类，使用keyedStream.timeWindow实现时间窗定义。

• Tumbling Time Window 滚动时间窗口：

/** 每1分钟，计算窗口数据 */ keyedStream.timeWindow(Time.minutes(1));

• Sliding Time Window 滑动时间窗口：

/** 每半分钟，计算最近1分钟窗口数据 */ keyedStream.timeWindow(Time.minutes(1), Time.seconds(30));

Flink Window组件

Flink Window使用3个组件协同实现了内置的3个窗口。通过对这3个组件不同的组合，可以满足许多场景的窗口定义。

WindowAssigner组件为数据分配窗口、Trigger组件决定如何处理窗口中的数据、借助Evictor组件实现灵活清理窗口中数据时机。

WindowAssigner

当有数据流入到Window Operator时需要按照一定规则将数据分配给窗口，WindowAssigner为数据分配窗口。下面代码片段是WindowAssigner部分定义，assignWindows方法定义返回的结果是一个集合，也就是说数据允许被分配到多个窗口中。

/*** WindowAssigner关键接口定义 ***/ public abstract class WindowAssigner implements Serializable { /** 分配数据到窗口集合并返回 */ public abstract Collection assignWindows(T element, long timestamp, WindowAssignerContext context); }

Flink内置WindowAssigner

Flink针对不同窗口类型实现了相应的WindowAssigner。Flink 1.7.0继承关系如下图

Trigger

Trigger触发器，它定义了3个触发动作，并且定义了触发动作处理完毕后的返回结果。返回结果交给Window Operator后由Window Operator决定后续操作。也就是说，Trigger通过具体的动作处理结果决定窗口是否应该被处理、被清除、被处理 清除、还是什么都不做。

/** Trigger关键接口定义 */ public abstract class Trigger implements Serializable { /*** 新的数据进入窗口时触发 ***/ public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception; /*** 处理时间计数器触发 ***/ public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception; /*** 事件时间计数器触发 ***/ public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception; }

当有数据流入Window Operator时会触发onElement方法、当处理时间和事件时间生效时会触发onProcessingTime和onEventTime方法。每个触发动作的返回结果用TriggerResult定义。

TriggerResult返回类型及说明

Trigger触发运算后返回处理结果，处理结果使用TriggerResult枚举表示。

public enum TriggerResult { CONTINUE,FIRE,PURGE,FIRE_AND_PURGE; }

Flink内置Trigger

Flink的内置窗口（Counter、Session、Time）有自己的触发器实现。下表为不同窗口使用的触发器。

Evictor

Evictor驱逐者，如果定义了Evictor当执行窗口处理前会删除窗口内指定数据再交给窗口处理，或等窗口执行处理后再删除窗口中指定数据。

public interface Evictor extends Serializable { /** 在窗口处理前删除数据 */ void evictBefore(Iterable