窗口分配器最通用的定义方式,就是调用.window()方法。这个方法需要传入一个 WindowAssigner 作为参数,返回 WindowedStream。如果是非按键分区窗口,那么直接调 用.windowAll()方法,同样传入一个 WindowAssigner,返回的是 AllWindowedStream。
1. 时间窗口
????????时间窗口是最常用的窗口类型,又可以细分为滚动、滑动和会话三种。? ????????标准的声明方式就是直接调用.window(),在里面传 入对应时间语义下的窗口分配器。我们不需要专门定义时间语义,默认就是事件时 间;如果想用处理时间,那么在这里传入处理时间的窗口分配器就可以了。?
(1)滚动处理时间窗口
窗口分配器由类 TumblingProcessingTimeWindows 提供,需要调用它的静态方法.of()。 ?
stream.keyBy(...)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.aggregate(...)
????????这里.of()方法需要传入一个 Time 类型的参数 size,表示滚动窗口的大小,我们这里创建 了一个长度为 5 秒的滚动窗口。? ????????另外,.of()还有一个重载方法,可以传入两个 Time 类型的参数:size 和 offset。第一个参 数当然还是窗口大小,第二个参数则表示窗口起始点的偏移量。这里需要多做一些解释:对于 我们之前的定义,滚动窗口其实只有一个 size 是不能唯一确定的。比如我们定义 1 天的滚动 窗口,从每天的 0 点开始计时是可以的,统计的就是一个自然日的所有数据;而如果从每天的 凌晨 2 点开始计时其实也完全没问题,只不过统计的数据变成了每天 2 点到第二天 2 点。这个 起始点的选取,其实对窗口本身的类型没有影响;而为了方便应用,默认的起始点时间戳是窗 口大小的整倍数。也就是说,如果我们定义 1 天的窗口,默认就从 0 点开始;如果定义 1 小时 的窗口,默认就从整点开始。而如果我们非要不从这个默认值开始,那就可以通过设置偏移量 offset 来调整。
(2)滑动处理时间窗口
????????窗口分配器由类 SlidingProcessingTimeWindows 提供,同样需要调用它的静态方法.of()。
stream.keyBy(...)
.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.aggregate(...)
????????这里.of()方法需要传入两个 Time 类型的参数:size 和 slide,前者表示滑动窗口的大小, 后者表示滑动窗口的滑动步长。我们这里创建了一个长度为 10 秒、滑动步长为 5 秒的滑动窗 口。 ????????滑动窗口同样可以追加第三个参数,用于指定窗口起始点的偏移量,用法与滚动窗口完全 一致。
(3)处理时间会话窗口
????????窗口分配器由类 ProcessingTimeSessionWindows 提供,需要调用它的静态方法.withGap()或者.withDynamicGap()。
stream.keyBy(...)
.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)))
.aggregate(...)
????????这里.withGap()方法需要传入一个 Time 类型的参数 size,表示会话的超时时间,也就是最 小间隔 session gap。我们这里创建了静态会话超时时间为 10 秒的会话窗口。
.window(ProcessingTimeSessionWindows.withDynamicGap(new
SessionWindowTimeGapExtractor<Tuple2<String, Long>>() {
@Override
public long extract(Tuple2<String, Long> element) {
// 提取 session gap 值返回, 单位毫秒
return element.f0.length() * 1000;
}
}))
????????这里.withDynamicGap()方法需要传入一个 SessionWindowTimeGapExtractor 作为参数,用 来定义 session gap 的动态提取逻辑。在这里,我们提取了数据元素的第一个字段,用它的长 度乘以 1000 作为会话超时的间隔。?
(4)滚动事件时间窗口
????????窗口分配器由类 TumblingEventTimeWindows 提供,用法与滚动处理事件窗口完全一致。
stream.keyBy(...)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.aggregate(...)
????????这里.of()方法也可以传入第二个参数 offset,用于设置窗口起始点的偏移量。
(5)滑动事件时间窗口
????????窗口分配器由类 SlidingEventTimeWindows 提供,用法与滑动处理事件窗口完全一致。
stream.keyBy(...)
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.aggregate(...)
(6)事件时间会话窗口
stream.keyBy(...)
.window(EventTimeSessionWindows.withGap(Time.seconds(10)))
.aggregate(...)
|