Flink项目中Window那些事

说明：本文为《Flink大数据项目实战》学习笔记，想通过视频系统学习Flink这个最火爆的大数据计算框架的同学，推荐学习CSDN官网课程：

Flink大数据项目实战：http://t.cn/ExrHPl9

 

啥是Window？有啥作用？

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

Window是一种切割无限数据集为有限块并进行相应计算的处理手段(跟keyBy一样，也是一种分组手段，只不过同一event可能被分到多个组)。

在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。在这种情况下，我们必须定义一个窗口，用来收集最近一分钟内的数据，并对这个窗口内的数据进行计算。

Window分类(是否kyBy决定了大分类)

Keyed Windows(在已经按照key分组的基础上(KeyedStream)，再构建多任务并行window)。

Non-Keyed Windows(在未分组的DataStream上构建单任务window，并行度是1，API都带All后缀)

1.3Keyed Windows vs Non-Keyed Windows(以基于time的window为例)

1.4Window的生命周期

创建：当属于该窗口的第一个元素到达时就会创建该窗口

 

销毁：当时间(event/process time)超过窗口的结束时间戳+用户指定的延迟时(allowedLateness())，窗口将被移除(仅限time-based window)

例如:对于一个每5分钟创建Tumbling Windows(即翻滚窗口)窗口，允许1分钟的时延，Flink将会在12:00到12:05这段时间内第一个元素到达时创建窗口，当watermark超过12:06时，该窗口将被移除

 

Trigger(触发器):指定了窗口函数在什么条件下可被触发，触发器还可以决定在创建和删除窗口之间的任何时间清除窗口的内容。在这种情况下，清除仅限于窗口中的元素，而不是窗口元数据。这意味着新数据仍然可以添加到该窗口中。

 例如：当窗口中的元素个数超过4个时“ 或者 ”当水印达到窗口的边界时“触发计算。

 

Window 的函数:函数里定义了应用于窗口(Window)内容的计算逻辑。

 

Evictor(驱逐者)：将在触发器触发之后或者在函数被应用前后，清除窗口中的元素。

2. 窗口分配器(Window Assingers)

 

2.1Window Assinger是干啥的

当你决定stream是否keyby之后，window是没有构建的，你还需要指定一个window Assinger用于定义元素如何分配到窗口中。

 

window Assinger如何指定？

1.Keyedstream：window(WindowAssigner)

2.non-keyed streams ：windowAll(WindowAssigner)

 

window Assinger的作用：负责将每个传入的元素分配给一个或多个窗口。

2.2Window小分类(Window Assinger类型)

有了window Assinger，才会创建出各种形式的window来覆盖我们所需的各种场景，所以不用过多关注window本身的分类，关注window Assinger的分类即可。

 

Count-based window:根据元素个数对数据流进行分组切片

1.Tumbling CountWindow

2.Sliding CountWindow

 

Time-based window :根据时间对数据流进行分组切片

1.Tumbling Window

2.Sliding Window

3.Session Window

 

注意：Time-based window [start，end)

2.3Tumbling Windows-翻滚窗口

定义：将数据依据固定的窗口长度对数据进行切片。

 

特点：

1.时间对齐

2.窗口长度固定

3.event无重叠

 

适用场景：BI统计(计算各个时间段的指标)。

 

对齐方式：默认是aligned with epoch(整点、整分、整秒等)，可以通过offset参数改变对齐方式。

2.4Sliding Windows-滑动窗口

定义：是固定窗口的更广义的一种形式。滑动窗口由固定的窗口长度和滑动间隔组成。

 

特点：

1.时间对齐

2.窗口长度固定

3.event有重叠

 

适用场景： 监控场景，对最近一个时间段内的统计(求某接口最近5min的失败率来决定是否要报警)。

对齐方式：默认是aligned with epoch(整点、整分、整秒等)，可以通过offset参数改变对齐方式。

2.5Session Windows

定义：类似于web应用 的session，即一段时间没有接受到新数据就会生成新的窗口(固定gap/gap fun)。

 

特点：

1.时间无对齐

2.event不重叠

3.没有固定开始和结束时间

 

适用场景： 线上用户行为分析。

 

Gap

1.固定gap

2.动态gap：实现SessionWindowTimeGapExtractor

 

特殊处理方式

1.session window operator为每个到达的event创建一个新窗口，如果它们之间的距离比定义的间隔更近，则将窗口合并在一起。

2.为了能够合并， session window operator需要合并触发器和合并窗口函数，例如ReduceFunction、AggregateFunction或ProcessWindowFunction

(FoldFunction不能合并)。

2.6Global Windows

定义：有相同key的所有元素分配给相同的单个全局窗口，必须指定自定义触发器否则没有任何意义。

注意：不要跟Non-keyed Window搞混，两个不同的角度。

 

使用方式：

 

2.7所有窗口盘点

2.8预定义Keyed Windows

Tumbling time window

.timeWindow(Time.seconds(30))

 

Sliding time window

.timeWindow(Time.seconds(30), Time.seconds(10))

 

Tumbling count window

.countWindow(1000)

 

Sliding count window

.countWindow(1000, 10)

 

Session window

.window(SessionWindows.withGap(Time.minutes(10)))

2.9预定义Non-Keyed Windows

Tumbling time window

.timeWindowAll(Time.seconds(30))

 

Sliding time window

.timeWindowAll(Time.seconds(30), Time.seconds(10))

 

Tumbling count window

.countWindowAll(1000)

 

Sliding count window

.countWindowAll(1000, 10)

 

Session window

.window(SessionWindows.withGap(Time.minutes(10)))

3. 窗口函数(作用在window上的Operator)

3.1盘点窗口函数们

在定义了窗口分配器之后,我们需要为每一个窗口明确的指定计算逻辑,这个就是窗口函数要做的事情,当系统决定一个窗口已经准备好执行之后，这个窗口函数将被用来处理窗口中的每一个元素(可能是分组的)。

 

谁可以作为窗口函数来使用：

特别提示：在没有专门说明的情况下，凡是带All的API就是给Non-keyed window使用的。

3.2ReduceFunction

含义：ReduceFunction定义了如何把两个输入的元素进行合并来生成相同类型的输出元素的过程，Flink使用ReduceFunction来对窗口中的元素进行增量聚合。

3.3AggregateFunction

AggregateFunction是ReduceFunction的普适版本，它需要指定三个类型:输入类型(IN)、累加器类型(ACC)和输出类型(OUT)。输入类型是输入流中的元素类型，AggregateFunction有一个方法可以将一个输入元素添加到一个累加器中。该接口还具有创建初始累加器、将两个累加器合并到一个累加器以及从累加器中提取输出(类型为OUT)的方法。

3.4FoldFunction

含义：FoldFunction指定了一个输入元素如何与一个指定输出类型的元素合并的过程，这个FoldFunction 会被每一个加入到窗口中的元素和当前的输出值增量地调用，第一个元素是与一个预定义的类型为输出类型的初始值合并。

3.5WindowFunction/AllWindowFunction(会逐步退出历史舞台)

含义：一个WindowFunction将获得一个包含了window中的所有元素迭代(Iterable)，并且提供灵活性。这些带来了性能的成本和资源的消耗，因为window中的元素无法进行增量迭代，而是缓存起来直到window被认为是可以处理时为止。

 

可以跟ReduceFunction /AggregateFunction/FoldFunction结合使用(推荐用法)。

3.6ProcessWindowFunction/ProcessAllWindowFunction

含义：ProcessWindowFunction获得一个包含窗口所有元素的可迭代器，以及一个具有时间和状态信息访问权的上下文对象，这使得它比其他窗口函数提供更大的灵活性。这是以性能和资源消耗为代价的，因为元素不能增量地聚合，而是需要在内部缓冲，直到认为窗口可以处理为止。

WindowFunctionde的升级版，可以跟ReduceFunction /AggregateFunction/FoldFunction结合使用(推荐用法)。

3.7ProcessWindowFunction/ProcessAllWindowFunction与ReduceFunction混搭

含义： ProcessWindowFunction可以与ReduceFunction、AggregateFunction或FoldFunction组合，以便在元素到达窗口时增量地聚合它们。当窗口关闭时，ProcessWindowFunction将提供聚合结果。ProcessWindowFunction可以在访问附加窗口元信息的同时进行增量计算。

3.8在ProcessWindowFunction中使用每个窗口的状态

含义： ProcessWindowFunction可以与ReduceFunction、AggregateFunction或FoldFunction组合，以便在元素到达窗口时增量地聚合它们。当窗口关闭时，ProcessWindowFunction将提供聚合结果。ProcessWindowFunction可以在访问附加窗口元信息的同时进行增量计算。

4. 触发器(Triggers)

4.1什么是触发器

触发器决定了一个窗口何时可以被窗口函数处理(条件满足时触发并发出信号)。

每一个WindowAssigner都有一个默认的触发器，如果默认的触发器不能满足你的需要，你可以通过调用trigger(...)来指定一个自定义的触发器。

 

触发器有5个方法来允许触发器处理不同的事件(Trigger):

1.onElement()方法,每个元素被添加到窗口时调用。

2.onEventTime()方法,当一个已注册的事件时间计时器启动时调用。

3.onProcessingTime()方法,当一个已注册的处理时间计时器启动时调用。

4.onMerge()方法，与状态性触发器相关，当使用session window时，两个触发器对应的窗口合并时，合并两个触发器的状态。

5.clear() 相应窗口被清除时触发。

 

前三个个方法通过返回TriggerResult来决定如何对其调用事件进行操作。该操作可以是以下操作之一：

1.CONTINUE:什么也不做

2.FIRE:触发计算

3.PURGE:清除窗口中的数据

4.FIRE_AND_PURGE:触发计算并随后清除窗口中的元素

这些方法中的任何一个都可以用于为将来的操作注册ProcessingTime计时器或EventTime计时器。

4.2触发和清除(Fire and Purge)

一旦一个触发器决定一个窗口已经准备好进行处理，它将触发并返回FIRE或者FIRE_AND_PURGE。这是窗口操作发送当前窗口结果的信号，发送给一个带有ProcessWindowFunction的窗口，所有元素都被传递给ProcessWindowFunction(可能在将它们传递给回收器之后)。

信号发送给具有ReduceFunction、AggregateFunction或FoldFunction的窗口只发出它们聚合的结果。

当一个触发器触发时，它可以是FIRE或者FIRE_AND_PURGE，如果是FIRE的话，将保持window中的内容，FIRE_AND_PURGE的话会清除window的内容。默认情况下，预实现的触发器仅仅是FIRE，不会清除window的状态。

注意:清除操作仅清除window的内容，并留下潜在的窗口元信息和完整的触发器状态。

4.3默认触发器

每一个窗口分配器都有一个默认的触发器。

WindowAssigner的默认触发器覆盖了很多场景(基本够用了)。例如，所有event-time window assigner都有一个EventTimeTrigger作为默认触发器。只要水印通过窗口的末端，这个触发器就会触发。

GlobalWindow的默认触发器是永不触发的NeverTrigger。因此在使用GlobalWindow时，必须定义一个自定义触发器。

通过使用trigger()指定触发器，将覆盖WindowAssigner的默认触发器。例如，如果你为TumblingEventTimeWindows指定了CountTrigger，那么将不再根据时间的进度获得窗口触发，而只根据计数。如果希望根据时间和计数进行响应，就必须编写自己的触发器。

4.4内置和自定义触发器

Flink内置的触发器：

1.EventTimeTrigger(前面提到过) 根据由watermark衡量的Event Time进度来触发。

2.ProcessingTimeTrigger 根据处理时间来触发。

3.CountTrigger 一旦窗口中的元素个数超出了给定的限制就会触发

4.PurgingTrigger 接受另一个触发器作为参数，并将其转换为一个purging触发器(当嵌套触发器触发时，将返回FIRE_AND_PURGE类型的TriggerResult)

 

自定义触发器

1.实现Trigger

2.还再发展，未来可能会有变化

5. 驱逐器(Evictors)

5.1驱逐器的作用

Evictor是可选的，WindowAssigner默认没有Evitor。

Evictor能够在Trigger触发之后以及在应用窗口函数执行前和/或后从窗口中删除无用的元素，类似filter作用(并不不是清除数据的)。

evictBefore()包含驱逐逻辑，在窗口函数之前应用。而evictAfter()在窗口函数之后应用。在应用窗口函数之前被逐出的元素将不被窗口函数处理。

5.2内置驱逐器

内置Evictor

1.CountEvitor：在窗口中保持用户指定数量的元素，并在窗口的开始处丢弃多余的元素

2.DeltaEvitor：通过一个DeltaFunction和一个阈值，计算窗口缓存中最后一个元素和剩余的所有元素的delta值，并清除delta值大于或者等于阈值的元素

3.TimeEvitor：使用一个interval(毫秒数)作为参数，对于一个给定的窗口，它会找出元素中的最大时间戳max_ts，并清除时间戳小于max_tx - interval的元素。

注意：

1.指定驱逐器可以防止任何预聚合，因为在应用计算之前，窗口的所有元素都必须传递给驱逐器。

2.Flink不能保证窗口中元素的顺序。这意味着，尽管驱逐者从窗口的开头删除元素，但这些元素不一定是最先或最后到达的元素。

6. 允许延迟(Allowed Lateness)

6.1如何允许延迟

当处理event-time的window时，可能会出现元素晚到的情况(即Flink用来跟踪event-time进度的watermark已经过了元素所属窗口的最后时间，属于当前窗口的数据才到达)。

1.默认情况下，当watermark已经过了窗口的最后时间时，晚到的元素会被丢弃。

2.Flink允许为窗口操作指定一个最大允许时延， Allowed lateness指定了元素可以晚到多长时间，默认情况下是0。

3.水位线已经过了窗口最后时间后才来的元素，如果还未到窗口最后时间加时延时间，那么元素任然添加到窗口中。根据所使用的触发器，延迟但未删除的元素可能导致窗口再次触发。例如使用EventTimeTrigger时。

4.特例：在使用GlobalWindows时，不会考虑延迟数据，因为全局窗口的结束时间戳是Long.MAX_VALUE。

6.2使用side output获取延迟的数据

当处理event-time的window时，可能会出现元素晚到的情况(即Flink用来跟踪event-time进度的watermark已经过了元素所属窗口的最后时间，属于当前窗口的数据才到达)。

6.3晚到元素注意事项

当指定允许的延迟大于0时，watermark通过窗口的末尾之后，窗口及其内容将保持不变。在这些情况下，当一个迟到但未删除的元素到达时，它可能触发窗口的另一个触发。

这些触发称为延迟触发，因为它们是由延迟事件触发的，而主触发是窗口的第一次触发。对于session window，延迟触发可能进一步导致窗口合并，因为它们可能“桥接”两个已存在的未合并窗口之间的差距。

特别注意：延迟触发发出的元素应该被视为以前计算的更新结果，即您的数据流将包含相同计算的多个结果。需要自行考虑【幂等性】问题。

 7.使用window result

7.1窗口化操作的结果

数据从WindowedStream/AllWindowedStream → （转换为）DataStream。

Window result结果元素中不保留关于窗口化操作的信息(可以在ProcessWindowFunction自己编码处理)。

Window result结果元素的timestamp=窗口允许的max timestamp = end timestamp – 1。

7.2Watermark与window的交互

watermark到达window operator时触发两件事：

1.当最大时间戳(end timestamp - 1)小于新watermark时， watermark将触发所有窗口的计算。

2.watermark原样转发给下游。

 

案例：连续的窗口操作

本实例首先根据key进行partition，然后再按指定的window对这些key进行计数，之后对该dataStream进行windowAll操作，其时间WindowAssigner与前面的相同，这样可以达到在同样的时间窗口内先partition汇总，再全局汇总的效果(可以解决类似top-k elements的问题)。

7.3估算窗口计算所需存储的注意事项

大窗口会累积很大的窗口状态，占用很大的存储(例如按月)。

估算窗口计算所需存储的注意事项：

1.滚动窗口(tumbling window)为每个元素保留一个副本(一个元素只属于一个窗口)； sliding window的每个元素有多个副本，因此不要创建size很大滑动步长很小的窗口(元素的副本数会膨胀)

2.ReduceFunction、AggregateFunction和FoldFunction可以显著降低存储需求，因为他们是增量聚合，且每个窗口只存储一个值。相反，仅仅使用processwindow函数就需要积累所有元素。

3.指定驱逐器可以防止任何预聚合，因为在应用计算之前，窗口的所有元素都必须传递给驱逐器。

7.4总结

Window Assigner ：决定某个元素被分配到哪个/哪些窗口中去。

Trigger ：触发器，触发窗口的计算或数据清除，每个Window Assigner有一个默认的Trigger。

Evictor ：“驱逐者”，类似filter作用。在Trigger触发之后，window被处理前或者后，Evictor用来删除窗口中无用的元素。默认没有驱逐器。