Flink中window 窗口和时间以及watermark水印

Posted 2022-02-04 Leo Han

我们都知道，Flink的核心是流式处理，但同时也支持批处理，Flink底层是一个流式引擎，在这个上面实现了流处理和批处理，而窗口则是批处理的实现。

在Flink中window从大的分类上主要有三种：Time Window（根据时间）、Count Window（根据数据量）、Session Window(会话窗口）

窗口类型有如下两种：

• Tumbling Window 滚动窗口,窗口之间的数据没有重叠
• Sliding Window 滑动窗口，窗口之间的数据有可能重叠
  

Count Window

Count Window主要有两类：

• Tumble Count Window:累积固定个数的数据就作为一个窗口，按照数据量来统计，不是按照时间，比如countWindow(100) 表示当窗口中的数据有100个的时候开始计算
• Sliding Count Window: 累积固定个数的数据就作为一个窗口，超过指定数量个数数据开启新一个窗口计算，比如 coutnWindow(100,10)窗口大小是100，滑动间隔是10，每增加10个元素就会对前面的100个元素计算一次

Time Window

Time Window是根据时间对数据进行分组的，

• Tumble Time Window: 在时间上按照给定的窗口大小切分窗口，窗口之间不会重叠
• Sliding Time Window: 在时间上按照给定的窗口大小、滑动步长切分窗口，窗口之间可能会存在数据重叠

Session Window

session window意为会话窗口，与HTTP请求的session概念类似，当超过一段时间，窗口没有收到数据时，认为窗口结束，计算窗口内的数据，窗口之间数据不会重叠

对于TimeWindow，在Flink中有几个时间语义：
Flink中主要有如下三种时间语义：

• Event Time ,数据自带的时间属性，使用这个语义时需要指定数据中哪个字段表示该时间同事必须设置WaterMark。使用Event Time时，数据可能是乱序的。在计算时，Flink会缓存窗口内的数据直到接收到WaterMark，WaterMark假设不会有更晚的数据到达，意味着在同一个时间窗口下，Flink会等待一个有限的时间，在一定程度上降低了计算结果的绝对准确性，并且增加了系统的延迟
• Processing Time： 数据进入某个算子，算子开始执行时的系统时间，不需要WaterMark机制，只依赖当前节点的操作系统时间
• Ingestion Time： 数据到达Flink Source的时间，从Source到下游的各个算子中可能有多个计算环节，任何一个算子处理速度的快慢可能影响下游算子的Processing Time,而 Ingestion Time定义的是数据流入Flink的时间，不会被下游算子处理速度影响，因此Ingestion Time通常是Event Time和Processing Time的一个折中方案，Ingestion Time不需要设置复Water Mark，也不需要太多缓存，延迟较低。

watermark水印

watermark一般是在Event Time语义下使用，我们知道，Event Time 是事件发生的时间，但是进入到Flink中并不一定按照EventTime顺序进入，导致窗口收到的时间会存在乱序问题，这种情况下，数据可能出现乱序和延迟情况，而WaterMark就是为了解决这个问题。
我们假设窗口为滚动窗口，窗口大小为1分钟，正常情况下，窗口如下：
窗口1 ： [20212-08 08:00:00 , 20212-08 08:01:00)
窗口2： [20212-08 08:01:00 , 20212-08 08:02:00)
窗口3： [20212-08 08:02:00 , 20212-08 08:03:00)

如果有一条数据的时间时间为 20212-08 08:00:30 在 20212-08 08:01:15到达，这个时候窗口1在20212-08 08:01:00已经处理完了，按照正常情况下，这条数据会被丢弃。
采用WaterMark机制之后，比如设置MaterMark为延迟20秒，那么这时候窗口1要在 20212-08 08:01:20的是才会触发计算，相当于这个窗口等了20秒之后才触发计算，而在等待20s的时间内，如果在[20212-08 08:00 , 20212-08 08:01:00)在这延迟的20s能够到达，那么也会纳入窗口1的计算中。

在Flink中如果时间语义设置为Event Time:env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);的话必须设置WaterMark。

Flink中WaterMark生成方式有两种：

• 周期性生成，一般每隔200ms生成一个WaterMark
• 数据流中每个数据eventTime都产生一个watermark