1. 窗口的概念
Flink是一种流式计算引擎,主要是来处理无界数据流,数据流的数据是一直都有的,等待流结束输入数据获取所有的流数据在做聚合计算是不可能的。为了更方便高效的处理无界流,一种方式就是把无限的流数据切割成有限的数据块
进行处理,这就是Flink中提到的窗口(Windows)
。
在Flink中,窗口就是用来处理无界流的核心。我们很容易把窗口想象成一个固定位置的框
,数据源源不断的流过来,到某个时间点窗口该关闭了,就停止收集数据,触发计算并输出结果。
例如,我们定义了一个时间窗口,每10秒统计一次数据,呢么就相当于把窗口放在那里,从0秒开始收集数据,到10秒时,处理当前窗口内所有的数据,输出一个结果,然后清空窗口继续收集数据;到20秒时,再对窗口内所有数据进行计算处理,输出结果;以此类推:
这里使用的窗口[0,10)窗口是左闭右开区间,即包含起始时间点,但不包括结束时间点。对于处理实时数据的窗口来说,这种方式存在一定问题。因为基于系统时间进行窗口关闭操作,在某些情况下可能会出现处理结果不准确或丢失部分数据的情况。例如,在一个 0-10 秒的窗口关闭后,如果还有一条时间戳为 9 秒的数据到达,则该数据将无法被正确地处理,并只能进入下一个 10-20 秒的窗口中。
然而如果我们采用事件时间语义,就会有一些费解了。由于乱序数据,我们需要设置一个延迟时间来等所有数据到齐。比如上面的例子,我们可以设置延迟时间为2秒,如下图,这样0-10秒的窗口会在时间戳为12秒的数据到来之后,才真正关闭计算输出结果,这样就可以正常包含迟到的9秒数据了。
但是这样一来,0-10秒的窗口不光包含了迟到的9秒数据,连11秒和12秒的数据也包含进去了。我们为了正确处理迟到数据,结果把早到的数据划分到了错误的窗口----最终结果也是错的
所以为了解决这个问题,窗口其实并不是一个框,流进来的数据被框住只能进这一个窗口。窗口而是一个桶。在Flink中,窗口可以把流切割成有限大小的多个存储桶;每个数据都会分发到对应的桶中,当到达窗口结束时间时,就对每个桶中收集的数据进行计算处理
在事件时间语义下,窗口的处理过程:
1. 第一个数据时间戳为2,判断之后创建第一个窗口[0,10),并将2秒数据保存进去;
2. 后续数据依次到来,时间戳均在[0,10)范围内,所以全部保存进第一个窗口
3. 11秒数据到来,判断不属于[0,10)窗口,所以创建第二个窗口[10,20),并将11秒的数据保存进去。由于水位线设置延迟时间为2秒,所以现在的时钟是9秒,第一个窗口也没有到关闭时间;
4. 之后又有9秒数据到来,同样进入[0,10)窗口中;
5. 12秒数据到来,判断属于[10,20)窗口,保存进去。这时产生的水位线推进到了10秒,所以[0,10)窗口应该关闭了。第一个 窗口收集到了所有的7个数据,进行处理计算后输出结果,并将窗口关闭销毁;
6. 同样的,之后的数据依次进入第二个窗口,遇到20秒的数据时会创建第三个窗口[20,30)并将数据保存进去;遇到22秒数据时,水位线到了20秒,第二个窗口触发计算,输出结果并关闭
注意!!! Flink 中窗口并不是静态准备好的,而是动态创建的——当有落在这个窗口区间范围的数据到达时,才创建对应的窗口。另外,这里我们认为到达窗口结束时间时,窗口就触发计算并关闭,事实上触发计算和窗口关闭两个行为也可以分开。
2. 窗口的分类
Flink中有很多种类的窗口,上面说的就是最简单的一种时间窗口
2.1 按照驱动类型分类
窗口本身是截取有界数据的一种方式,所以窗口最重要的信息就是怎样截取数据
,以什么标准来开始和结束数据的截取,叫做窗口的驱动类型
。
2.1.1 时间窗口(Time Window)
时间窗口(Time Window)
就是按照时间段去截取数据,这也是最常见的窗口。时间窗口以时间点来定义窗口的开始(start)
和结束(end)
,所以截取出的就是某一时间段的数据。到达结束时间时,窗口不再收集数据,触发计算输出结果,并将窗口关闭销毁,也可以说基本思路就是定点发车
。
用结束时间减去开始时间,得到这段时间的长度,就是窗口的大小(windows size)。这里的时间可以是不同的语义,所以我们可以定义处理时间窗口和事件时间窗口。
Flink中有一个专门的类来表示时间窗口,名称叫做TimeWindow
。这个类只有两个私有属性start
和end
,这表示窗口的开始和结束的时间戳,单位为毫秒。可以通过公有的get方法调用。另外TImeWindow
还提供了一个maxTimestamp()
方法,用来获取窗口中能够包含数据的最大时间戳。通过代码可以看出最大时间戳就是end-1
,这也代表了时间窗口的时间范围都是左闭右开的区间[start,end)
。
@PublicEvolving
public class TimeWindow extends Window {
private final long start;
private final long end;
public TimeWindow(long start, long end) {
this.start = start;
this.end = end;
}
public long getStart() {
return start;
}
public long getEnd() {
return end;
}
@Override
public long maxTimestamp() {
return end - 1;
}
....
}
2.1.2 计数窗口(CountWindow)
计数窗口是基于元素个数来截取数据,到达固定的个数时就触发计算并关闭窗口。类似于座位有限,坐满就发车,至于是否发车和时间没有任何关系。每个窗口的截取数据的个数,就是窗口的大小。
计数窗口相比时间窗口就更加简单,我们只需要指定窗口大小,就可以把数据分配到对应的窗口中,在Flink中没有相对应的类表示计数窗口,底层通过全局窗口(Global Window)来实现的。maxTimestamp
返回的Long.MAX_VALUE
@PublicEvolving
public class GlobalWindow extends Window {
private static final GlobalWindow INSTANCE = new GlobalWindow();
private GlobalWindow() {}
public static GlobalWindow get() {
return INSTANCE;
}
@Override
public long maxTimestamp() {
return Long.MAX_VALUE;
}
@Override
public boolean equals(Object o) {
return this == o || !(o == null || getClass() != o.getClass());
}
@Override
public int hashCode() {
return 0;
}
@Override
public String toString() {
return "GlobalWindow";
}
....
}
2.2 按照窗口分配数据的规则分类
时间窗口和计数窗口只是对窗口的一个大致划分,再具体应用时,还需要定义更加精细的规则,来控制数据应该划分到哪个窗口。不同的分配数据的方式,就可以有不同的功能应用。
根据分配数据的规则,窗口的具体实现划分为4类:滚动窗口(Tumbling Window)、滑动窗口(Sliding Window)、会话窗口(Session Window)、以及全局窗口(Global Window)。
2.2.1 滚动窗口(Tumbling Windows)
滚动窗口有固定的大小,是一种对数据进行的均匀切片的划分方式。窗口之间没有重叠,也不会有间隔,是"均匀切片"的划分你方式。窗口之间没有重叠,也不会相隔,是首尾相接的状态。如果我们把多个窗口的创建,看作一个窗口的运动,就类似于在不停的向前翻滚一样。这是最简单的窗口形式。也因为滚动窗口是无缝衔接,所以每个数据都会被分配到一个窗口上,而且也只属于一个窗口。
滚动窗口可以基于时间定义,也可以基于数据个数定义;需要的参数只有一个:窗口的大小(Windows Size)。窗口的大小可以使一个小时一次,也可以是长度为10的数据个数。
如上图所示,圆点表示数据流的数据,对数据按照userID做了分区。当固定了窗口大小之后,所有的分区的窗口划分都是一致的;窗口没有重叠,每个数据只属于一个窗口。
滚动窗口应用非常广泛,它可以对每个时间段做聚合统计,很多BI分析指标都可以用它来实现。
2.2.2 滑动窗口(Sliding Windows)
滑动窗口和滚动窗口类似,滑动窗口的大小也是很固定的。区别在于窗口之间并不是首尾相连的,而是错开一定的位置。如果看作一个窗口的运动,呢么就像是向前小步滑动一样,所以滑动窗口的参数就有两个,一个是窗口大小(Windows Size),一个是滑动的步长(Windows slide),它其实就代表了窗口计算的频率。滑动的距离代表了下个窗口开始的时间间隔,而窗口大小是固定的,所以也就是两个窗口结束时间的间隔;窗口在结束时间触发计算输出结果,呢么滑动步长就代表了计算频率。例如:我们定义一个长度为1小时,滑动步长为5分钟的滑动窗口,呢么就会统计1小时内的数据,每5分钟统计一次。同样,滑动窗口也可以基于时间定义,也可以基于数据个数定义。
当滑动步长小于窗口大小时,滑动窗口就会出现重叠,这时候的部分数据也可能被同时分配到多个窗口中去。而具体的个数,就由窗口大小和滑动步长的比值(size/slide)来决定。如图6-18所示,滑动步长刚好是窗口大小的一半,呢么在windows1和windows2的中间部分,每个数据都会被分配到这2个窗口里。。比如窗口长度定义1个小时,滑动步长为30分钟,呢么对于8.55的数据就分别属于[8,9)和[8.30,9.30]这两个窗口;
所以,滑动窗口是固定大小窗口的更广义的一种形式;换句话说,滚动窗口也是一种特殊的滑动窗口——窗口大小等于滑动步长(size==slide)
2.2.3 会话窗口(Session Windows)
会话窗口是基于会话(session)来对数据进行分组的。这里的会话类似Web的会话session概念,不过并不代表两端的通讯过程,而是借用会话超时失效的机制来描述窗口。简单来说就是当有数据来了就开启一个窗口,如果还有数据到来就一直保持开启状态,如果在等待一段时间后没有收到数据,就认为会话失效窗口自动关闭。
与滑动窗口和滚动窗口不同,会话窗口只能基于时间来定义,而没有"会话计数窗口"的概念。类似于"会话"终止的标志就是"隔一段时间没有数据来",如果不依赖时间而改成个数,就成了"隔几个数据没有来",这是自相矛盾的说法。
会话窗口有两个重要概念,一个是这段时间的长度——Size,它表示会话的超时时间,也就是两个会话窗口之间的最小距离。还有一个是两个数据到来的时间间隔——Gap,如果新的数据到来时间小于指定的大小size,那说明还在保持会话,就属于同一个窗口;但如果gap大于size,呢么新来的数据就应该属于新的会话窗口,前一个窗口就需要关闭了。具体实现上还可以设置静态固定大小Size,也可以通过一个自定义提取器(Gap Extractor)动态提取最小间隔Gap的值
考虑到事件时间语义下的乱序流,这里又会有一些麻烦。相邻两个数据的时间间隔 gap
大于指定的 size,我们认为它们属于两个会话窗口,前一个窗口就关闭;可在数据乱序的情况
下,可能会有迟到数据,它的时间戳刚好是在之前的两个数据之间的。这样一来,之前我们判
断的间隔中就不是“一直没有数据”,而缩小后的间隔有可能会比 size 还要小——这代表三个
数据本来应该属于同一个会话窗口。所以在 Flink 底层,对会话窗口的处理会比较特殊:每来一个新的数据,都会创建一个新的会话窗口;然后判断已有窗口之间的距离,如果小于给定的 size,就对它们进行合并(merge)操作。在 Window 算子中,对会话窗口会有单独的处理逻辑。
会话窗口和之前两种窗口不同,没有固定长度,起始和结束时间也不确定,各个分区之间窗口也是没有联系的。如图 6-19 所示,会话窗口之间一定是不会重叠的,而且会留有至少为 size 的间隔(session gap)。
2.2.4 全局窗口(Global Windows)
还有一类比较通用的窗口,就是“全局窗口”。这种窗口全局有效,会把相同 key 的所有数据都分配到同一个窗口中;说直白一点,就跟没分窗口一样。无界流的数据永无止尽,所以这种窗口也没有结束的时候,默认是不会做触发计算的。如果希望它能对数据进行计算处理,还需要自定义“触发器”(Trigger)。
如图 6-20 所示,可以看到,全局窗口没有结束的时间点,所以一般在希望做更加灵活的窗口处理时自定义使用。Flink 中的计数窗口(Count Window),底层就是用全局窗口实现的。