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
- 測試數據