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