OVER視窗函數使用文法及執行個體 - Realtime Compute for Apache Flink

OVER視窗（OVER Window）是傳統資料庫的標準開窗，不同於Group By Window，OVER視窗中每1個元素都對應1個視窗。OVER視窗可以按照實際元素的行或實際的元素值（時間戳記值）確定視窗，因此流資料元素可能分布在多個視窗中。

在應用OVER視窗的流式資料中，每1個元素都對應1個OVER視窗。每1個元素都觸發1次資料計算，每個觸發計算的元素所確定的行，都是該元素所在視窗的最後1行。在Realtime Compute的底層實現中，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個商品中的最高價格。

測試表tmall_item資料

itemid(VARCHAR)	itemtype(VARCHAR)	eventtime(VARCHAR)	price(DOUBLE)
ITEM001	Electronic	`2024-11-11 10:01:00`	20
ITEM002	Electronic	`2024-11-11 10:02:00`	50
ITEM003	Electronic	`2024-11-11 10:03:00`	30
ITEM004	Electronic	`2024-11-11 10:03:00`	60
ITEM005	Electronic	`2024-11-11 10:05:00`	40
ITEM006	Electronic	`2024-11-11 10:06:00`	20
ITEM007	Electronic	`2024-11-11 10:07:00`	70
ITEM008	Clothes	`2024-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 '2' SECOND  -- 為Rowtime定義Watermark
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<brokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

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	`2024-11-11 10:01:00`	20	20
ITEM002	Electronic	`2024-11-11 10:02:00`	50	50
ITEM003	Electronic	`2024-11-11 10:03:00`	30	50
ITEM004	Electronic	`2024-11-11 10:03:00`	60	60
ITEM005	Electronic	`2024-11-11 10:05:00`	40	60
ITEM006	Electronic	`2024-11-11 10:06:00`	20	60
ITEM007	Electronic	`2024-11-11 10:07:00`	70	70
ITEM008	Clothes	`2024-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分鐘的同類商品中的最高價格。

測試表tmall_item資料

itemid(VARCHAR)	itemtype(VARCHAR)	eventtime(VARCHAR)	price(DOUBLE)
ITEM001	Electronic	`2024-11-11 10:01:00`	20
ITEM002	Electronic	`2024-11-11 10:02:00`	50
ITEM003	Electronic	`2024-11-11 10:03:00`	30
ITEM004	Electronic	`2024-11-11 10:03:00`	60
ITEM005	Electronic	`2024-11-11 10:05:00`	40
ITEM006	Electronic	`2024-11-11 10:06:00`	20
ITEM007	Electronic	`2024-11-11 10:07:00`	70
ITEM008	Clothes	`2024-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 '2' SECOND  -- 為Rowtime定義Watermark
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<brokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

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	`2024-11-11 10:01:00`	20	20
ITEM002	Electronic	`2024-11-11 10:02:00`	50	50
ITEM003	Electronic	`2024-11-11 10:03:00`	30	50
ITEM004	Electronic	`2024-11-11 10:03:00`	60	60
ITEM005	Electronic	`2024-11-11 10:05:00`	40	60
ITEM006	Electronic	`2024-11-11 10:06:00`	20	40
ITEM007	Electronic	`2024-11-11 10:07:00`	70	70
ITEM008	Clothes	`2024-11-11 10:08:00`	20	20