窗口是flink处理无限流的核心,窗口将流拆分为有限大小的“桶”，我们可以在这些桶上进行计算。
1、Keyed vs Non-Keyed Windows
根据上游数据是否为Keyed Stream类型(是否将数据按照某个指定的Key进行分区)，将窗口划分为Keyed Window和Non-Keyed Windows。两者的区别在于KeyStream调用相应的window()方法来指定window类型，数据会根据Key在不同的Task中并行计算，而Non-Keyed Stream需要调用WindowsAll()方法来指定window类型，所有的数据都会在一个Task进行计算，相当于没有并行。
1.1 Keyed Windows
stream
.keyBy(...) <- keyed versus non-keyed windows
.window(...) <- 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"
1.2 Non-Keyed Windows
stream
.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"
2、窗口分配器
窗口分配器负责将一个事件分配给一个或多个窗口，内置窗口包括： 滚动窗口（Tumbling Windows）、滑动窗口（Sliding Windows）、会话窗口（Session Windows）、全局窗口（Global Windows），也可以通过继承WindowAssigner类来自定义窗口。
2.1 基于时间的窗口
Flink中所有的内置窗口（全局窗口除外）都有基于时间的实现，这个时间可以是事件时间（event time），也可以是处理时间（processing time）。其中，处理滚动窗口和滑动窗口的算子，在1.12版本之前使用timeWindow()，在1.12版本被标记为废弃，转而使用window()来作为窗口处理算子，这里只介绍最新版本的使用算子。
-
滚动时间窗口（Tumbling Time Windows）
滚动窗口将每一个事件分配给一个有特定大小的窗口，滚动窗口有固定大小，不会重叠。比如一个滚动窗口大小（window size）为5分钟。
使用示例如下：
DataStream<T> input = ...;
// tumbling event-time windows
input
.keyBy(<key selector>)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.<windowed transformation>(<window function>);
// tumbling processing-time windows
input
.keyBy(<key selector>)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.<windowed transformation>(<window function>);
// 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>);
由于Flink默认使用的时间基准是UTC±00:00时间，在中国需要使用UTC+08:00时间，所以最后一个示例中窗口大小为1天，时间偏移量就是8小时。
-
滑动窗口（Sliding Time Windows）
跟滚动窗口类似，滑动窗口也是将每一个事件分配给特定大小的窗口，且窗口有固定的大小，但它有一个窗口滑动的参数，标识一个窗口滑动的频率，或者说是每隔多久窗口滑动一次。比如一个窗口的大小为10秒钟，滑动频率为5秒。
使用示例如下：
DataStream<T> input = ...;
// sliding event-time windows
input
.keyBy(<key selector>)
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.<windowed transformation>(<window function>);
// sliding processing-time windows
input
.keyBy(<key selector>)
.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.<windowed transformation>(<window function>);
// 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>);
最后一个示例，中的Time.hours(-8)含义与滚动窗口一致。从滑动窗口的使用来看，滚动窗口其实是滑动窗口的一个特例，但窗口大小和滑动间隔相等的时候，滑动窗口就是一个滚动窗口。
-
会话窗口（Session Windows）
会话窗口按活动的会话对事件进行分组。与滑动窗口和滚动窗口相比，会话窗口没有固定的大小，也没有固定的起止时间，它是以一段时间没有接收到事件为窗口结束条件的。会话窗口分配器可以配置成一个固定的session gap，或者定义成一个session gap提取函数，在函数中定义一个不活跃的时长，一旦这个时长结束，当前会话就结束。
使用示例如下：
DataStream<T> input = ...;
// event-time session windows with static gap
input
.keyBy(<key selector>)
.window(EventTimeSessionWindows.withGap(Time.minutes(10)))
.<windowed transformation>(<window function>);
// event-time session windows with dynamic gap
input
.keyBy(<key selector>)
.window(EventTimeSessionWindows.withDynamicGap((element) -> {
// determine and return session gap
}))
.<windowed transformation>(<window function>);
// processing-time session windows with static gap
input
.keyBy(<key selector>)
.window(ProcessingTimeSessionWindows.withGap(Time.minutes(10)))
.<windowed transformation>(<window function>);
// processing-time session windows with dynamic gap
input
.keyBy(<key selector>)
.window(ProcessingTimeSessionWindows.withDynamicGap((element) -> {
// determine and return session gap
}))
.<windowed transformation>(<window function>);
动态的会话gap需要实现SessionWindowTimeGapExtractor接口。
2.2 基于计数的窗口
基于计数的窗口是根据事件的个数来对窗口进行划分的，概念跟基于时间的滚动窗口差不多，只不过窗口大小的划分，有时间变成了事件的个数。
- 滚动计数窗口（Tumbling Count Windows）
stream
.keyBy(1)
.countWindow(100) \\100为事件的个数,即窗口的大小
.sum(1);
- 滑动计数窗口（Sliding Count Windows）
stream
.keyBy(1)
.countWindow(100, 10) \\100为事件的个数，即窗口的大小，10为滑动的间隔
.sum(1);
2.3 全局窗口（Global Windows）
全局窗口分配器将所有具有相同key的元素分配到同一个全局窗口中，这个窗口模式仅适用于用户还需自定义触发器的情况。否则，由于全局窗口没有一个自然的结尾，无法执行元素的聚合，将不会有计算被执行。
使用示例如下：
DataStream<T> input = ...;
input
.keyBy(<key selector>)
.window(GlobalWindows.create())
.<windowed transformation>(<window function>);
3、触发器(Triggers)
触发器决定了一个窗口何时可以被窗口函数处理，每一个窗口分配器都有一个默认的触发器，如果默认的触发器不能满足你的需要，你可以通过调用trigger(...)来指定一个自定义的触发器。触发器的接口有5个方法来允许触发器处理不同的事件:
- onElement()方法,每个元素被添加到窗口时调用
- onEventTime()方法,当一个已注册的事件时间计时器启动时调用
- onProcessingTime()方法,当一个已注册的处理时间计时器启动时调用
- onMerge()方法，与状态性触发器相关，当使用会话窗口时，两个触发器对应的窗口合并时，合并两个触发器的状态。
- 最后一个clear()方法执行任何需要清除的相应窗口
Flink有一些内置的触发器:
- EventTimeTrigger(前面提到过)触发是根据由水印衡量的事件时间的进度来的
- ProcessingTimeTrigger 根据处理时间来触发
- CountTrigger 一旦窗口中的元素个数超出了给定的限制就会触发
- PurgingTrigger 作为另一个触发器的参数并将它转换成一个清除类型
如果你想实现一个自定义的触发器需要继承Trigger类
GlobalWindow默认的触发器是NeverTrigger，是永远不会触发的，因此，如果你使用的是GlobalWindow的话，需要定义一个自定义触发器。
4、驱逐器(Evictors)
Flink的窗口模型允许指定一个除了WindowAssigner和Trigger之外的可选参数Evitor，这个可以通过调用evitor(...)方法来实现。这个驱逐器(evitor)可以在触发器触发之前或者之后清理窗口中的元素。为了达到这个目的，Evitor接口有两个方法:
void evictBefore(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
void evictAfter(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
注：指定一个Evitor要防止预聚合，因为窗口中的所有元素必须得在计算之前传递到驱逐器中
网友评论