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

在應用OVER視窗的流式資料中，每1個元素都對應1個OVER視窗。每1個元素都觸發1次資料計算，每個觸發計算的元素所確定的行，都是該元素所在視窗的最後1行。在實時計算的底層實現中，OVER視窗的資料進行全域性統一管理（資料只儲存1份），邏輯上為每1個元素維護1個OVER視窗，為每1個元素進行視窗計算，完成計算後會清除過期的資料。

語法

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的每個元素都確定一個視窗。ROWS OVER Window分為Unbounded（無界流）和Bounded（有界流）兩種情況。

  Unbounded ROWS OVER Window資料示例如下圖所示。

說明雖然上圖所示視窗user1的w7、w8及user2的視窗w3、w4都是同一時刻到達，但它們仍然在不同的視窗，這一點與RANGE OVER Window不同。 Bounded ROWS OVER Window資料以3個元素（往前2個元素）的視窗為例，如下圖所示。

說明雖然上圖所示視窗user1的w5、w6及user2的視窗w1、w2都是同一時刻到達，但它們仍然在不同的視窗，這一點與RANGE 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 TABLE tmall_item(
   itemID VARCHAR,
   itemType VARCHAR,
   onSellTime TIMESTAMP,
   price DOUBLE,
   WATERMARK onSellTime FOR onSellTime as withOffset(onSellTime, 0)
) 
WITH (
  type = '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所有具有共同元素值（元素時間戳）的元素行確定一個視窗，RANGE OVER Window分為Unbounded和Bounded的兩種情況。

  Unbounded RANGE OVER Window資料示例如下圖所示。

說明上圖所示視窗user1的w7、user2的視窗w3，兩個元素同一時刻到達，屬於相同的window，這一點與ROWS OVER Window不同。 Bounded RANGE OVER Window資料，以3秒中資料(INTERVAL '2' SECOND)的視窗為例，如下圖所示。

• 說明上圖所示視窗user1的w6、user2的視窗w3，元素都是同一時刻到達，屬於相同的window，這一點與ROWS 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 TABLE tmall_item(
   itemID VARCHAR,
   itemType VARCHAR,
   onSellTime TIMESTAMP,
   price DOUBLE,
   WATERMARK onSellTime FOR onSellTime as withOffset(onSellTime, 0)
) 
WITH (
  type = '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