本文为您介绍如何使用Flink滚动窗口函数。
定义
滚动窗口(TUMBLE)将每个元素分配到一个指定大小的窗口中。通常,滚动窗口有一个固定的大小,并且不会出现重叠。例如,如果指定了一个5分钟大小的滚动窗口,无限流的数据会根据时间划分为[0:00, 0:05)
、[0:05, 0:10)
、[0:10, 0:15)
等窗口。
语法
标识函数
使用标识函数选出窗口的起始时间或者结束时间,窗口的时间属性用于下级Window的聚合。
窗口标识函数 | 返回类型 | 描述 |
---|---|---|
TUMBLE_START(time-attr, size-interval) |
TIMESTAMP | 返回窗口的起始时间(包含边界)。例如[00:10,00:15) 窗口,返回00:10 。
|
TUMBLE_END(time-attr, size-interval) |
TIMESTAMP | 返回窗口的结束时间(包含边界)。例如[00:00, 00:15] 窗口,返回00:15 。
|
TUMBLE_ROWTIME(time-attr, size-interval) |
TIMESTAMP(rowtime-attr) | 返回窗口的结束时间(不包含边界)。例如(00:00, 00:15) 窗口,返回00:14:59.999 。返回值是一个rowtime attribute,即可以基于该字段进行时间属性的操作,例如,级联窗口只能用在基于Event Time的Window上,详情请参见级联窗口。
|
TUMBLE_PROCTIME(time-attr, size-interval) |
TIMESTAMP(rowtime-attr) | 返回窗口的结束时间(不包含边界)。例如(00:00, 00:15) 窗口,返回00:14:59.999 。返回值是一个Proctime Attribute,即可以基于该字段进行时间属性的操作。例如,级联窗口只能用在基于Processing Time的Window上,详情请参见级联窗口。
|
使用Event Time统计每个用户每分钟在指定网站的单击数示例
- 测试数据
username(VARCHAR) click_url(VARCHAR) eventtime(VARCHAR) Jark http://taobao.com/xxx
2017-10-10 10:00:00.0
Jark http://taobao.com/xxx
2017-10-10 10:00:10.0
Jark http://taobao.com/xxx
2017-10-10 10:00:49.0
Jark http://taobao.com/xxx
2017-10-10 10:01:05.0
Jark http://taobao.com/xxx
2017-10-10 10:01:58.0
Timo http://taobao.com/xxx
2017-10-10 10:02:10.0
- 测试语句
CREATE TEMPORARY TABLE user_clicks( username varchar, click_url varchar, eventtime varchar, ts AS TO_TIMESTAMP(eventtime), WATERMARK FOR ts AS ts - INTERVAL '2' SECOND --为Rowtime定义Watermark。 ) WITH ( 'connector'='sls', ... ); CREATE TEMPORARY TABLE tumble_output( window_start TIMESTAMP, window_end TIMESTAMP, username VARCHAR, clicks BIGINT ) WITH ( 'connector'='datahub' --目前SLS只支持输出VARCHAR类型的DDL,所以使用DataHub存储。 ... ); INSERT INTO tumble_output SELECT TUMBLE_START(ts, INTERVAL '1' MINUTE) as window_start, TUMBLE_END(ts, INTERVAL '1' MINUTE) as window_end, username, COUNT(click_url) FROM user_clicks GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE),username;
- 测试结果
window_start (TIMESTAMP) window_end (TIMESTAMP) username(VARCHAR) clicks(BIGINT) 2017-10-10 10:00:00.0
2017-10-10 10:01:00.0
Jark 3 2017-10-10 10:01:00.0
2017-10-10 10:02:00.0
Jark 2 2017-10-10 10:02:00.0
2017-10-10 10:03:00.0
Timo 1
使用Processing Time统计每个用户每分钟在指定网站的单击数示例
- 测试数据
username (VARCHAR) click_url(VARCHAR) Jark http://taobao.com/xxx
Jark http://taobao.com/xxx
Jark http://taobao.com/xxx
Jark http://taobao.com/xxx
Jark http://taobao.com/xxx
Timo http://taobao.com/xxx
- 测试语句
CREATE TEMPORARY TABLE window_test ( username VARCHAR, click_url VARCHAR, ts as PROCTIME() ) WITH ( 'connector'='sls', ... ); CREATE TEMPORARY TABLE tumble_output( window_start TIMESTAMP, window_end TIMESTAMP, username VARCHAR, clicks BIGINT ) WITH ( 'connector'='datahub' --目前SLS只支持输出VARCHAR类型的DDL,所以使用DataHub存储。 ... ); INSERT INTO tumble_output SELECT TUMBLE_START(ts, INTERVAL '1' MINUTE), TUMBLE_END(ts, INTERVAL '1' MINUTE), username, COUNT(click_url) FROM window_test GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), username;
- 测试结果
window_start (TIMESTAMP) window_end (TIMESTAMP) username (VARCHAR) clicks(BIGINT) 2019-04-11 14:43:00.000
2019-04-11 14:44:00.000
Jark 5 2019-04-11 14:43:00.000
2019-04-11 14:44:00.000
Timo 1