OVER窗口(OVER Window)是传统数据库的标准开窗,不同于Group By Window,OVER窗口中每1个元素都对应1个窗口。OVER窗口可以按照实际元素的行或实际的元素值(时间戳值)确定窗口,因此流数据元素可能分布在多个窗口中。
在应用OVER窗口的流式数据中,每1个元素都对应1个OVER窗口。每1个元素都触发1次数据计算,每个触发计算的元素所确定的行,都是该元素所在窗口的最后1行。在实时计算的底层实现中,OVER窗口的数据进行全局统一管理(数据只存储1份),逻辑上为每1个元素维护1个OVER窗口,为每1个元素进行窗口计算,完成计算后会清除过期的数据。详情请参见Over Aggregation。
语法
SELECT
agg1(col1) OVER (definition1) AS colName,
...
aggN(colN) OVER (definition1) AS colNameN
FROM Tab1;
- agg1(col1):按照GROUP BY指定col1列对输入数据进行聚合计算。
- OVER (definition1):OVER窗口定义。
- AS colName:别名。
说明
- agg1到aggN所对应的OVER definition1必须相同。
- 外层SQL可以通过AS的别名查询数据。
类型
Flink SQL中对OVER窗口的定义遵循标准SQL的定义语法,传统OVER窗口没有对其进行更细粒度的窗口类型命名划分。按照计算行的定义方式,OVER Window可以分为以下两类:
- ROWS OVER Window:每1行元素都被视为新的计算行,即每1行都是一个新的窗口。
- RANGE OVER Window:具有相同时间值的所有元素行视为同一计算行,即具有相同时间值的所有行都是同一个窗口。
属性
正交属性 | 说明 | proctime | eventtime |
---|---|---|---|
ROWS OVER Window | 按照实际元素的行确定窗口。 | 支持 | 支持 |
RANGE OVER Window | 按照实际的元素值(时间戳值)确定窗口。 | 支持 | 支持 |
Rows OVER Window语义
- 窗口数据
ROWS OVER Window的每个元素都确定一个窗口。
- 窗口语法
SELECT agg1(col1) OVER( [PARTITION BY (value_expression1,..., value_expressionN)] ORDER BY timeCol ROWS BETWEEN (UNBOUNDED | rowCount) PRECEDING AND CURRENT ROW) AS colName, ... FROM Tab1;
- value_expression:分区值表达式。
- timeCol:元素排序的时间字段。
- rowCount:定义根据当前行开始向前追溯几行元素。
- 案例
以Bounded ROWS OVER Window场景为例。假设,一张商品上架表,包含有商品ID、商品类型、商品上架时间、商品价格数据。要求输出在当前商品上架之前同类的3个商品中的最高价格。
- 测试数据
商品ID 商品类型 上架时间 销售价格 ITEM001 Electronic 2017-11-11 10:01:00
20 ITEM002 Electronic 2017-11-11 10:02:00
50 ITEM003 Electronic 2017-11-11 10:03:00
30 ITEM004 Electronic 2017-11-11 10:03:00
60 ITEM005 Electronic 2017-11-11 10:05:00
40 ITEM006 Electronic 2017-11-11 10:06:00
20 ITEM007 Electronic 2017-11-11 10:07:00
70 ITEM008 Clothes 2017-11-11 10:08:00
20 - 测试代码
CREATE TEMPORARY TABLE tmall_item( itemID VARCHAR, itemType VARCHAR, eventtime varchar, onSellTime AS TO_TIMESTAMP(eventtime), price DOUBLE, WATERMARK FOR onSellTime AS onSellTime - INTERVAL '0' SECOND --为Rowtime定义Watermark。 ) WITH ( 'connector' = 'sls', ... ); SELECT itemID, itemType, onSellTime, price, MAX(price) OVER ( PARTITION BY itemType ORDER BY onSellTime ROWS BETWEEN 2 preceding AND CURRENT ROW) AS maxPrice FROM tmall_item;
- 测试结果
itemID itemType onSellTime price maxPrice ITEM001 Electronic 2017-11-11 10:01:00
20 20 ITEM002 Electronic 2017-11-11 10:02:00
50 50 ITEM003 Electronic 2017-11-11 10:03:00
30 50 ITEM004 Electronic 2017-11-11 10:03:00
60 60 ITEM005 Electronic 2017-11-11 10:05:00
40 60 ITEM006 Electronic 2017-11-11 10:06:00
20 60 ITEM007 Electronic 2017-11-11 10:07:00
70 70 ITEM008 Clothes 2017-11-11 10:08:00
20 20
- 测试数据
RANGE OVER Window语义
- 窗口数据
RANGE OVER Window所有具有共同元素值(元素时间戳)的元素行确定一个窗口。
- 窗口语法
SELECT agg1(col1) OVER( [PARTITION BY (value_expression1,..., value_expressionN)] ORDER BY timeCol RANGE BETWEEN (UNBOUNDED | timeInterval) PRECEDING AND CURRENT ROW) AS colName, ... FROM Tab1;
- value_expression:进行分区的字表达式。
- timeCol:元素排序的时间字段。
- timeInterval:定义根据当前行开始向前追溯指定时间的元素行。
- 案例
Bounded RANGE OVER Window场景示例:假设一张商品上架表,包含有商品ID、商品类型、商品上架时间、商品价格数据。需要求比当前商品上架时间早2分钟的同类商品中的最高价格。
- 测试数据
商品ID 商品类型 上架时间 销售价格 ITEM001 Electronic 2017-11-11 10:01:00
20 ITEM002 Electronic 2017-11-11 10:02:00
50 ITEM003 Electronic 2017-11-11 10:03:00
30 ITEM004 Electronic 2017-11-11 10:03:00
60 ITEM005 Electronic 2017-11-11 10:05:00
40 ITEM006 Electronic 2017-11-11 10:06:00
20 ITEM007 Electronic 2017-11-11 10:07:00
70 ITEM008 Clothes 2017-11-11 10:08:00
20 - 测试代码
CREATE TEMPORARY TABLE tmall_item( itemID VARCHAR, itemType VARCHAR, eventtime varchar, onSellTime AS TO_TIMESTAMP(eventtime), price DOUBLE, WATERMARK FOR onSellTime AS onSellTime - INTERVAL '0' SECOND --为Rowtime定义Watermark。 ) WITH ( 'connector' = 'sls', ... ); SELECT itemID, itemType, onSellTime, price, MAX(price) OVER ( PARTITION BY itemType ORDER BY onSellTime RANGE BETWEEN INTERVAL '2' MINUTE preceding AND CURRENT ROW) AS maxPrice FROM tmall_item;
- 测试结果
itemID itemType onSellTime price maxPrice ITEM001 Electronic 2017-11-11 10:01:00
20 20 ITEM002 Electronic 2017-11-11 10:02:00
50 50 ITEM003 Electronic 2017-11-11 10:03:00
30 50 ITEM004 Electronic 2017-11-11 10:03:00
60 60 ITEM005 Electronic 2017-11-11 10:05:00
40 60 ITEM006 Electronic 2017-11-11 10:06:00
20 40 ITEM007 Electronic 2017-11-11 10:07:00
70 70 ITEM008 Clothes 2017-11-11 10:08:00
20 20
- 测试数据