流计算 Oceanus 时间窗口函数

窗口函数是一种特殊的函数，它并不在 SELECT 的投影列表中使用，而是在 GROUP BY 子句中使用。流计算 Oceanus 支持三种类型的窗口函数 TUMBLE、HOP 和 SESSION。

Flink 流处理介绍详见 及时流处理。

一、TUMBLE WINDOW

TUMBLE WINDOW（滚动窗口）将每个进入的数据分配到一个指定窗口大小的窗口中。滚动窗口可以自定义固定的大小，并且不会出现重叠。我们可以对窗口内的数据进行计算。

二、语法

TUMBLE(time_attr, interval)

·time_attr 参数表示时间戳字段，表示每条记录被处理的时间戳。如果指定为 PROCTIME 是自动生成的时间戳，记录了数据被 flink 处理的时刻，一般用在 Processing Time 模式下。

·interval 参数用来设置窗口大小。例如，设置为1天：INTERVAL '1' DAY；设置为2小时：INTERVAL '2' HOUR，其他用法可参见 时间相关函数。

三、标识函数

标识函数用来表示窗口的起始与结束时间。

四、模拟用例

下文以 TUMBLE WINDOW 为例，帮助你更容易地理解 TUMBLE WINDOW。使用 Event Time 模拟统计每小时各用户收入金额。

示例数据：

SQL 语句：

CREATE TABLE user_income ( username VARCHAR, Income INT, times TIMESTAMP(3), WATERMARK FOR times AS times – INTERVAL '3' SECOND) WITH ( … );CREATE TABLE output ( win_start TIMESTAMP, win_end TIMESTAMP, username VARCHAR, hour_income BIGINT)WITH( …);INSERT INTO outputSELECTTUMBLE_START(times,INTERVAL '1' HOUR),TUMBLE_END(times,INTERVAL '1' HOUR),username,SUM(Income)FROM user_incomeGROUP BY TUMBLE(times,INTERVAL '1' HOUR),username;

输出结果：

五、HOP WINDOW

HOP WINDOW（滑动窗口）将元素分配到固定长度的窗口中，与滚动窗口类似，窗口的大小由窗口大小参数来配置，另一个窗口滑动参数控制滑动窗口开始的频率。

HOP WINDOW（滑动窗口）保持窗口大小（Size）不变，每次滑动指定的时间周期（Slide），因而允许窗口之间的相互重叠。

Slide 的大小决定了 Flink 创建新窗口的频率。

·当 Slide 小于 Size 时（如图 window1 与 window2），相邻窗口会重叠，一个时间会被分配到多个窗口。

·当 Slide 大于 Size 时（如图 window1 与 window4），可能会导致有些事件被丢弃。

·当 Slide 等于 Size 时（如图 window1 与 window3），等于是 TUMBLE WINDOW。

六、语法

HOP(time_attr, sliding_interval, window_size_interval)

·time_attr 参数表示时间戳字段，表示每条记录被处理的时间戳。如果指定为 PROCTIME 是自动生成的时间戳，记录了数据被 flink 处理的时刻，一般用在 Processing Time 模式下。

·window_size_interval 参数用来设置窗口大小。例如，设置为1天：INTERVAL '1' DAY；设置为2小时：INTERVAL '2' HOUR，其他用法可参见 时间相关函数。

·sliding_interval 参数用来设置滑动时间周期大小。例如，设置为1天：INTERVAL '1' DAY；设置为2小时：INTERVAL '2' HOUR，其他用法可参见 时间相关函数。

七、标识函数

标识函数用来表示窗口的起始与结束时间。

八、模拟用例

下文以 HOP WINDOW 为例，帮助你更容易地理解 HOP WINDOW。使用 Event Time 模拟统计每小时各用户收入金额，每30分钟更新一次。1小时的窗口，10分钟滑动一次。

样例数据：

SQL 语句：

CREATE TABLE user_income ( username VARCHAR, Income INT, times TIMESTAMP(3), WATERMARK FOR times AS times – INTERVAL '3' MINUTE)WITH( … );CREATE TABLE output ( win_start TIMESTAMP, win_end TIMESTAMP, username VARCHAR, hour_income BIGINT)WITH( … );INSERT INTO outputSELECTHOP_START(times,INTERVAL '30' MINUTE,INTERVAL '1' HOUR),HOP_END(times,INTERVAL '30' MINUTE,INTERVAL '1' HOUR),username,SUM(income)FROM user_incomeGROUP BY HOP(times,INTERVAL '30' MINUTE,INTERVAL '1' HOUR),username;

输出结果：

九、SESSION WINDOW

SESSION WINDOW（会话窗口）通过 session 活动对元素进行分组，session 窗口与滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况，相反，当它在一个固定的时间周期内不再收到元素，即非活动间隔产生，那么这个窗口就会关闭。一个 session 窗口通过一个 session 间隔来配置。这个 session 间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的 session 将关闭并且后续的元素将被分配到新的 session 窗口中。

Session Window 并非以长度来划分窗口，而是以非活跃时间来划分。例如超过30分钟不活跃（没有新数据），则之前的窗口结束，下一个来到的数据将会形成一个新窗口。

十、语法

SESSION(time_attr, interval)

·time_attr 参数表示时间戳字段，表示每条记录被处理的时间戳。如果指定为 PROCTIME 是自动生成的时间戳，记录了数据被 flink 处理的时刻，一般用在 Processing Time 模式下。

·interval 参数用来设置窗口大小。例如，设置为1天：INTERVAL '1' DAY；设置为2小时：INTERVAL '2' HOUR，其他用法可参见 时间相关函数。

十一、标识函数

标识函数用来表示窗口的起始与结束时间。

十二、模拟用例

下文以 SESSION WINDOW 为例，帮助你更容易地理解 SESSION WINDOW。使用 Event Time 模拟统计每小时各用户收入金额，会话超时时长为30分钟。

样例数据：

SQL 语句：

CREATE TABLE user_income ( username VARCHAR, Income INT, times TIMESTAMP(3), WATERMARK FOR times AS times – INTERVAL '3' MINUTE)WITH( … );CREATE TABLE output ( win_start TIMESTAMP, win_end TIMESTAMP, username VARCHAR, hour_income BIGINT)WITH( … );INSERT INTO outputSELECTSESSION_START(times,INTERVAL '30' MINUTE),SESSION_END(times,INTERVAL '30' MINUTE),username,SUM(Income)FORM user_incomeGROUP BY SESSION(times,INTERVAL '30' MINUTE),username;

输出结果：

十三、更多说明

以上三种窗口都有对应的辅助函数。以 TUMBLE 窗口为例（HOP、SESSION 也一样，只是前缀不同），辅助函数如下：

·TUMBLE_ROWTIME：表示 TUMBLE 窗口的末端界限（包含，可用作 JOIN 或 GROUP 以及 OVER 条件，Event Time 时间模式下使用）。示例如下：

SELECT user,TUMBLE_START(rowtime, INTERVAL '12' HOUR) AS sStart,TUMBLE_ROWTIME(rowtime, INTERVAL '12' HOUR) AS snd,SUM(amount)FROM OrdersGROUP BY TUMBLE(rowtime, INTERVAL '12' HOUR), user

·TUMBLE_PROCTIME：表示 TUMBLE 窗口的末端界限（包含，可用作 JOIN 或 GROUP 以及 OVER 条件，Processing Time 时间模式下使用）。示例如下：

SELECT user,TUMBLE_START(PROCTIME, INTERVAL '12' HOUR) AS sStart,TUMBLE_PROCTIME(PROCTIME, INTERVAL '12' HOUR) AS snd,SUM(amount)FROM OrdersGROUP BY TUMBLE(PROCTIME, INTERVAL '12' HOUR), user