4.1.17 Flink-流处理框架-Window API之窗口分配器

港控/mmm° 2022-11-05 00:39 152阅读 0赞

**目录**

0.写在前面

1.TimeWindow-时间窗口

1.1 滚动时间窗口（Tumbling Windows）

1.2 滑动时间窗口（SlidingEventTimeWindows）

1.3 三个时间窗口总结

2.CountWindow-计数窗口

2.1 滚动窗口

2.2 滑动窗口

2.3 两个计数窗口总结

--------------------

# 0.写在前面 #

我们可以用 .window() 来定义一个窗口，然后基于这个 window 去做一些聚 合或者其它处理操作。注意 window () 方法必须在 keyBy 之后才能用。

*  **想要调用.window()方法（窗口分配器），前面必须要先调用keyBy()方法**
 *  **调用完.window()方法之后，后面必须跟上一个聚合的窗口操作，我们称为窗口函数window function（下一节涉及到）**

Flink 提供了更加简单的 **.timeWindow** 和 **.countWindow** 方法，用于定义**时间窗口**和**计数窗口**。

window() 方法接收的输入参数是一个 WindowAssigner，WindowAssigner 负责将每条输入的数据分发到正确的 window 中。

Flink 提供了通用的 WindowAssigner：**➢ 滚动窗口（tumbling window） ➢ 滑动窗口（sliding window） ➢ 会话窗口（session window） ➢ 全局窗口（global window）**

# 1.TimeWindow-时间窗口 #

TimeWindow是将指定时间范围内的所有数据组成一个window，一次对一个window里面的所有数据进行计算。

## 1.1 滚动时间窗口（Tumbling Windows） ##

Flink **默认的时间窗口，**根据 Processing Time 进行窗口的划分，将 Flink 获取到的数据根据进入 Flink 的时间划分到不同的窗口中。

DataStream<Tuple2<String, Double>> minTempPerWindowStream = dataStream.map(new MapFunction<SensorReading, Tuple2<String, Double>>() {
        @Override
        public Tuple2<String, Double> map(SensorReading value) throws Exception {
            return new Tuple2<>(value.getId(), value.getTemperature());
        }
    }).keyBy(data -> data.f0).timeWindow( Time.seconds(15)).minBy(1);

时间间隔可以通过 Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其 中的一个来指定。

## 1.2 滑动时间窗口（SlidingEventTimeWindows） ##

滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是 window\_size，一个是 sliding\_size。

下面代码中的 sliding\_size 设置为了 5s，也就是说，每 5s 就计算输出结果一次， 每一次计算的 window 范围是 15s 内的所有元素。

DataStream<SensorReading> minTempPerWindowStream = dataStream
        .keyBy(SensorReading::getId)
        .timeWindow( Time.seconds(15), Time.seconds(5) )
        .minBy("temperature");

时间间隔可以通过 Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其 中的一个来指定。

## 1.3 三个时间窗口总结 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

# 2.CountWindow-计数窗口 #

CountWindow 根据窗口中相同 key 元素的数量来触发执行，执行时只计算元素 数量达到窗口大小的 key 对应的结果。

注意：CountWindow 的 window\_size 指的是相同 Key 的元素的个数，不是输入的所有元素的总数。

## 2.1 滚动窗口 ##

默认的 CountWindow 是一个滚动窗口，只需要指定窗口大小即可，当元素数量达到窗口大小时，就会触发窗口的执行。

DataStream<SensorReading> minTempPerWindowStream = dataStream
        .keyBy(SensorReading::getId)
        .countWindow( 5 )
        .minBy("temperature");

## 2.2 滑动窗口 ##

滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是 window\_size，一个是 sliding\_size。

下面代码中的 sliding\_size 设置为了 2，也就是说，每收到两个相同 key 的数据 就计算一次，每一次计算的 window 范围是 10 个元素。

DataStream<SensorReading> minTempPerWindowStream = dataStream
        .keyBy(SensorReading::getId)
        .countWindow( 10, 2 )
        .minBy("temperature");

## 2.3 两个计数窗口总结 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/ba9efebef31a410a8429cfeba571fc02.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/95379900faa34e9a9e3de9cdb8f9229f.png