Flink解析 | Time & Window

一.Window & Time介绍

Apache Flink(以下简称 Flink)是一个天然支持无限流数据处理的分布式计算框架，在Flink中Window可以将无限流切分成有限流，是处理有限流的核心组件，现在Flink中Window可以是时间驱动的(Time Window)，也可以是数据驱动的(Count Window)。

下面的代码是在Flink中使用Window的两个示例：

二.Window API使用

从第一部分我们已经知道Window的一些基本概念，以及相关API，下面我们以一个实际例子来看看怎么使用Window相关的API。

代码来自flink-examples

上面的例子中我们首先会对每条数据进行时间抽取，然后进行keyby，接着依次调用window(),evictor(),trigger()以及maxBy()。下面我们重点来看window(), evictor() 和trigger()这几个方法。

2.1 WindowAssigner, Evictor以及Trigger

window()方法接收的输入是一个WindowAssigner，WindowAssigner负责将每条输入的数据分发到正确的window中(一条数据可能同时分发到多个 Window 中)，Flink提供了几种通用的WindowAssigner：tumbling window(窗口间的元素无重复)，sliding window(窗口间的元素可能重复)，session window以及global window。如果需要自己定制数据分发策略，则可以实现一个class，继承自WindowAssigner。

Tumbling Window

Sliding Window

Session Window

Global Window

evictor()主要用于做一些数据的自定义操作，可以在执行用户代码之前，也可以在执行用户代码之后，更详细的描述可以参考org.apache.flink.streaming

.api.windowing.evictors.Evictor的evicBefore和evicAfter两个方法。Flink 提供了如下三种通用的 evictor：

• CountEvictor保留指定数量的元素

• DeltaEvictor通过执行用户给定的DeltaFunction以及预设的threshold，判断是否删除一个元素。

• TimeEvictor设定一个阈值interval，删除所有不再max_ts – interval范围内的元素，其中max_ts是窗口内时间戳的最大值。

evictor() 是可选的方法，如果用户不选择，则默认没有。

trigger()用来判断一个窗口是否需要被触发，每个WindowAssigner都自带一个默认的trigger，如果默认的trigger不能满足你的需求，则可以自定义一个类，继承自Trigger即可，我们详细描述下Trigger的接口以及含义：

• onElement()：每次往window增加一个元素的时候都会触发

• onEventTime()：当event-time timer被触发的时候会调用

• onProcessingTime()：当processing-time timer被触发的时候会调用

• onMerge()：对两个trigger的state进行merge操作

• clear()：window销毁的时候被调用

上面的接口中前三个会返回一个TriggerResult，TriggerResult有如下几种可能的选择：

• CONTINUE：不做任何事情

• FIRE：触发window

• PURGE：清空整个window的元素并销毁窗口

• FIRE_AND_PURGE：触发窗口，然后销毁窗口

2.2 Time & Watermark

了解完上面的内容后，对于时间驱动的窗口，我们还有两个概念需要澄清：Time和Watermark。

我们知道在分布式环境中Time是一个很重要的概念，在Flink中Time可以分为三种Event-Time，Processing-Time以及Ingestion-Time，三者的关系我们可以从下图中得知：

Event Time、Ingestion Time、Processing Time

• Event-Time：表示事件发生的时间

• Processing-Time：表示处理消息的时间(墙上时间)

• Ingestion-Time ： 表示进入到系统的时间。

在Flink中我们可以通过下面的方式进行Time类型的设置：

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); // 设置使用 ProcessingTime

了解了Time之后，我们还需要知道Watermark相关的概念。

我们可以考虑一个这样的例子：某App会记录用户的所有点击行为，并回传日志(在网络不好的情况下，先保存在本地，延后回传)。A用户在11:02对App进行操作，B 用户在11:03操作了App，但是A用户的网络不太稳定，回传日志延迟了，导致我们在服务端先接受到B用户11:03的消息，然后再接受到A用户11:02的消息，消息乱序了。

那我们怎么保证基于event-time的窗口在销毁的时候，已经处理完了所有的数据呢？这就是watermark的功能所在。watermark会携带一个单调递增的时间戳t，watermark(t)表示所有时间戳不大于t的数据都已经到来了，未来小于等于t的数据不会再来，因此可以放心地触发和销毁窗口了。下图中给了一个乱序数据流中的watermark例子：

2.3 迟到的数据

上面的watermark让我们能够应对乱序的数据，但是真实世界中我们没法得到一个完美的watermark数值—要么没法获取到，要么耗费太大，因此实际工作中我们会使用近似watermark—生成watermark(t)之后，还有较小的概率接受到时间戳t之前的数据，在Flink中将这些数据定义为“late elements”, 同样我们可以在window中指定是允许延迟的最大时间(默认为 0)，可以使用下面的代码进行设置：

设置 `allowedLateness`之后，迟来的数据同样可以触发窗口，进行输出，利用Flink 的side output机制，我们可以获取到这些迟到的数据，使用方式如下：

需要注意的是，设置了allowedLateness之后，迟到的数据也可能触发窗口，对于 Session window来说，可能会对窗口进行合并，产生预期外的行为。

3 Window 内部实现

在讨论Window内部实现的时候，我们再通过下图回顾一下Window的生命周期：

每条数据过来之后，会由WindowAssigner分配到对应的Window，当Window被触发之后，会交给 Evictor(如果没有设置 Evictor 则跳过)，然后处理UserFunction。其中WindowAssigner，Trigger，Evictor我们都在上面讨论过，而UserFunction则是用户编写的代码。

整个流程还有一个问题需要讨论：Window中的状态存储。我们知道Flink是支持Exactly Once处理语义的，那么Window中的状态存储和普通的状态存储又有什么不一样的地方呢？

首先给出具体的答案：从接口上可以认为没有区别，但是每个Window会属于不同的namespace，而非Window场景下，则都属于VoidNamespace，最终由State/Checkpoint来保证数据的Exactly Once语义，下面我们从org.apache.flink.streaming.runtime.operators.windowing.WindowOperator摘取一段代码进行阐述

从上面我们可以知道，Window中的的元素同样是通过state进行维护，然后由Checkpoint机制保证Exactly Once语义。

至此，Time、Window相关的所有内容都已经讲解完毕，主要包括为什么要有Window；Window中的三个核心组件：WindowAssigner、Trigger和Evictor；Window中怎么处理乱序数据，乱序数据是否允许延迟，以及怎么处理迟到的数据；最后我们梳理整个Window的数据流程，以及Window中怎么保证Exactly Once语义。

大家工作学习遇到HBase技术问题，把问题发布到HBase技术社区论坛http://hbase.group，欢迎大家论坛上面提问留言讨论。想了解更多HBase技术关注HBase技术社区公众号(微信号:hbasegroup)，非常欢迎大家积极投稿。

本群为HBase+Spark技术交流讨论，整合最优质的专家资源和技术资料会定期开展线下技术沙龙，专家技术直播，专家答疑活动

点击链接钉钉入群 ： https://dwz.cn/Fvqv066s 或扫码进群

本群为Cassandra技术交流讨论，整合最优质的专家资源和技术资料会定期开展线下技术沙龙，专家技术直播，专家答疑活动

Cassandra 社区钉钉大群： https://c.tb.cn/F3.ZRTY0o

Cassandra 技术社区微信公众号：