flink Window Assigners是Flink流处理中用于将数据流划分为有限大小的时间窗口的机制之一。Window Assigners负责将数据流中的数据分配到不同的时间窗口中。时间窗口可以基于不同的标准进行划分,如时间戳、事件数量或其他自定义标准。
Flink提供了多个Window Assigners实现,包括滚动窗口(Tumbling Windows)、滑动窗口(Sliding Windows)、会话窗口(Session Windows)等。滚动窗口是基于固定大小的时间段来划分数据流的,而滑动窗口则是根据固定大小和滑动间隔来划分数据流的。会话窗口根据一定的时间间隔来划分数据流,并根据数据之间的间隔来决定何时关闭会话窗口。
使用Window Assigners,可以将数据流中的数据分配到不同的时间窗口中,从而方便地进行聚合、计算和分析。
一、窗口(Window)
窗口(Window)是处理无界流的关键所在。窗口可以将数据流装入大小有限的“桶”中,再对每个“桶”加以处理。
1. 滚动窗口(Tumbling Windows)
滚动窗口有固定的大小,是一种对数据进行均匀切片的划分方式。窗口之间没有重叠,也不会有间隔,是“首尾相接”的状态。滚动窗口可以基于时间定义,也可以基于数据个数定义;需要的参数只有一个,就是窗口的大小(window size)。

2. 滑动窗口(Sliding Windows)
与滚动窗口类似,滑动窗口的大小也是固定的。区别在于,窗口之间并不是首尾相接的,而是可以“错开”一定的位置。如果看作一个窗口的运动,那么就像是向前小步“滑动”一样。定义滑动窗口的参数有两个:除去窗口大小(window size)之外,还有一个滑动步长(window slide),代表窗口计算的频率。

3. 会话窗口(Session Windows)
借用会话超时失效的机制来描述窗口简单来说,就是数据来了之后就开启一个会话窗口,如果接下来还有数据陆续到来,那么就一直保持会话;如果一段时间一直没收到数据,那就认为会话超时失效,窗口自动关闭。与滑动窗口和滚动窗口不同,会话窗口只能基于时间来定义,而没有“会话计数窗口”的概念。
考虑到事件时间语义下的乱序流,在 Flink 底层,对会话窗口的处理会比较特殊:每来一个新的数据,都会创建一个新的会话窗口;然后判断已有窗口之间的距离,如果小于给定的 size,就对它们进行合并(merge)操作。在 Window 算子中,对会话窗口会有单独的处理逻辑。

4. 全局窗口(Global Windows)
这种窗口全局有效,会把相同 key 的所有数据都分配到同一个窗口中;说直白一点,就跟没分窗口一样。无界流的数据永无止尽,所以这种窗口也没有结束的时候,默认是不会做触发计算的。如果希望它能对数据进行计算处理,还需要自定义触发器(Trigger)

二、Keyed Streams和Non-Keyed Streams
Flink 窗口在 keyed streams 和 non-keyed streams 上使用的基本结构。 我们可以看到,这两者唯一的区别仅在于:keyed streams 要调用 keyBy(…)后再调用 window(…) , 而 non-keyed streams 只用直接调用 windowAll(…)。
在Apache Flink中,数据流可以被分为两种类型:Keyed Streams和Non-Keyed Streams。
Keyed Streams是指在处理数据流时需要对数据进行分区,并将数据按照某个Key进行分组的数据流。Keyed Streams中的每个元素都有一个Key,通过这个Key来确定数据所属的分区。在Flink中,Keyed Streams通常用于聚合操作,如GroupBy和Window。
相反,Non-Keyed Streams是指在处理数据流时不需要按照某个Key进行分组的数据流。在Non-Keyed Streams中,每个元素都是相互独立的,没有固定的分区和分组关系。Non-Keyed Streams通常用于一些基于时间的操作,如TimeWindow和ProcessFunction。
区分Keyed Streams和Non-Keyed Streams的方法是,当定义数据流时,如果调用了DataStream的keyBy()方法,那么这个数据流就是Keyed Streams。如果没有调用keyBy()方法,则这个数据流就是Non-Keyed Streams。
import org.apache.flink.streaming.api.scala._
val env = StreamExecutionEnvironment.getExecutionEnvironment
val stream: DataStream[(String, Int)] = env
.socketTextStream("localhost", 9999)
.map { value =>
val Array(key, count) = value.split(",")
(key, count.toInt)
}
.keyBy(_._1) // 调用keyBy方法,将数据按照Tuple2的第一个元素进行分组,因此这个数据流是Keyed Streams
注意的是,在Scala中,对于Tuple类型的数据,可以直接使用元组的语法来访问元素。例如,可以使用_1来访问Tuple的第一个元素,使用2来访问第二个元素。因此,在上面的示例中,keyBy(._1)可以理解为按照Tuple的第一个元素进行分组。
import org.apache.flink.streaming.api.scala._
val env = StreamExecutionEnvironment.getExecutionEnvironment
val stream: DataStream[String] = env
.socketTextStream("localhost", 9999)
.flatMap { value =>
value.split(" ")
} // 没有调用keyBy方法,因此这个数据流是Non-Keyed Streams