視窗彙總 - Realtime Compute for Apache Flink

背景資訊

分組視窗彙總（老文法）：對應GroupWindowAggregation運算元，支援TUMBLE、HOP、SESSION視窗類別型。
視窗資料表值函式彙總（新文法）：基於Window TVF新文法的視窗彙總，具有所有效能調優中提到的效能最佳化措施、支援標準的GROUPING SETS文法、可以在視窗彙總結果上使用視窗Top等優勢。對應WindowAggregate運算元，支援TUMBLE、HOP、CUMULATE和SESSION視窗函數。

說明

分組視窗彙總已淘汰，推薦您使用更高效且功能更豐富的視窗資料表值函式彙總。
它們對於更新流的支援情況，請參見新老文法對更新流的支援情況。

分組視窗彙總（老文法）

分組視窗彙總定義在SQL的GROUP BY子句中，和普通的GROUP BY子句一樣，包含分組視窗函數的GROUP BY子句的查詢會對各組分別計算，各自產生一個結果行。

分組視窗彙總的文法、範例及特性等詳情，請參見Group Window Aggregation。

VVR 11.x 分組視窗彙總行為變更

從 VVR 11.x（對應 Flink 1.20）開始，系統不再預設將分組視窗彙總（老文法）自動改寫為視窗資料表值函式彙總（新文法）的執行計畫。

在 VVR 8.x 中，系統自動將老文法改寫為新文法的執行計畫，使其獲得 Local-Global 兩階段彙總最佳化。從 VVR 11.x 起，該自動改寫不再是預設行為，老文法保持原有的物理執行計畫。

影響：使用老文法的作業不再自動獲得兩階段彙總最佳化，在資料量大或資料扭曲情境下可能影響效能。

重要

推薦將分組視窗彙總遷移到視窗資料表值函式彙總（新文法）。關於兩階段彙總最佳化的詳細生效條件，請參見視窗彙總 Local-Global 最佳化。

相容舊版本行為

如暫時無法遷移，可手動開啟以下參數恢複 VVR 8.x 行為：

參數	說明	預設值
`table.optimizer.window-rewrite-enabled`	開啟老文法到新文法的自動改寫，以獲得兩階段彙總最佳化。	false（VVR 11.x 起）

配置樣本（在作業參數中添加）：

table.optimizer.window-rewrite-enabled: true

重要

此參數僅作為升級過渡的相容手段。老文法已標記為過時，後續版本可能不再維護改寫能力，建議儘早遷移到新文法。

視窗資料表值函式彙總（新文法）

視窗彙總是通過GROUP BY子句定義的，其特徵是包含由視窗資料表值函式產生的window_start和 window_end列。和普通的GROUP BY子句一樣，視窗彙總會為每個組計算出一行資料。

和其他連續表上的彙總不同，視窗彙總不產生中間結果，只在視窗結束產生一個總的彙總結果，另外，視窗彙總會清除不需要的中間狀態。

視窗資料表值函式彙總的文法、範例及特性等，請參見Window TVF Aggregation。

SESSION視窗資料表值函式彙總在不同VVR版本中的區別

VVR 11.x（對應Flink 1.20版本）文法

SESSION(TABLE data [PARTITION BY(keycols, ...)], DESCRIPTOR(timecol), gap)

參數含義如下：

data：擁有時間屬性列的表。
keycols：列描述符，決定會話視窗應該使用哪些列來分區資料。
timecol：列描述符，決定資料的哪個時間屬性列應該映射到視窗。
gap：兩個事件被認為屬於同一個會話視窗的最大時間間隔。

（建議棄用）VVR 8.x（對應Flink 1.17版本）文法

SESSION(TABLE data, DESCRIPTOR(timecol), gap)

參數含義如下：

data：擁有時間屬性列的表。
timecol：列描述符，決定資料的哪個時間屬性列應該映射到視窗。
gap：兩個事件被認為屬於同一個會話視窗的最大時間間隔。

Realtime Compute引擎VVR 8.x與VVR 11.x的SESSION視窗資料表值函式使用區別如下：

特性	VVR 11.x	VVR 8.x（建議棄用）	差異說明
文法結構	`SESSION(TABLE data [PARTITION BY(keycols, ...)], DESCRIPTOR(timecol), gap)`	`SESSION(TABLE data, DESCRIPTOR(timecol), gap)`	VVR 8.x缺少`PARTITION BY` 文法，需通過彙總語句隱式指定分區欄位。
分區欄位指定方式	支援顯式 `PARTITION BY(keycols)` 指定分區欄位。	必須通過彙總語句的 `GROUP BY` 子句隱式指定分區欄位。	VVR 8.x強制要求分區欄位為彙總語句中非 `window_start、window_end、window_time` 的GROUP KEY。
分區欄位限制	無強制限制，可自由選擇分區欄位。	分區欄位必須與彙總語句的 `GROUP BY` 欄位一致（且排除視窗時間欄位）。	VVR 8.x通過彙總邏輯隱式綁定分區欄位，VVR 11.x文法更靈活。
參數完整性	完整參數：`data`, `PARTITION BY`, `DESCRIPTOR(timecol)`, `gap`。	簡化參數：`data`, `DESCRIPTOR(timecol)`, `gap`。	VVR 8.x移除了 `PARTITION BY` 參數，依賴彙總上下文推斷分區邏輯。
單獨使用支援性	支援單獨調用 `SESSION()` 函數（不依賴彙總）。	不支援單獨使用，必須與彙總語句（如 `GROUP BY`）綁定。	VVR 8.x強制要求函數與彙總語句耦合，VVR 11.x支援更靈活的使用情境。
視窗函數與彙總合并性	支援視窗函數與彙總語句合并（如 `SUM(price)`）。	不支援視窗資料表值函式和彙總語句無法合并的情境（需保持彙總邏輯與視窗函數一致）。	VVR 8.x對彙總與視窗的合并使用有嚴格限制。

以下樣本中SQL文法是等價的，都將使用item欄位作為SESSION視窗函數的分區欄位：

-- tables must have time attribute, e.g. `bidtime` in this table
> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

-- VVR 11.x
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;
  
-- VVR 8.x
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

樣本對比說明：

情境	VVR 11.x SQL	VVR 8.x SQL（建議棄用）	說明
SESSION視窗分區	`SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)`	`SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)`	兩者均通過 `item` 欄位分區，但VVR 8.x需依賴 `GROUP BY item` 隱式綁定分區邏輯。
彙總與視窗合并	支援直接合并（如 `SUM(price)` 在視窗內計算）。	僅當彙總欄位與視窗分區欄位一致時支援（如 `GROUP BY item`）。	VVR 8.x對彙總與視窗的合并使用有隱性約束。

視窗資料表值函式和彙總語句無法合并的情境

以下情境均以SESSION視窗為例，同樣適用於其他視窗資料表值函式。

警告

當視窗資料表值函式和彙總語句無法合并時，如果使用Processing Time作為視窗列來劃分視窗，則會導致視窗彙總（Window Aggregation）節點使用被視窗資料表值函式（Window TVF）節點物化的Processing Time列作為視窗的時間屬性，在彙總計算時會受到來自於源表浮水印（Watermark）的幹擾，從而導致視窗提前輸出並可能出現類似事件時間視窗的延遲資料丟棄。請改寫您的SQL，盡量避免視窗資料表值函式和彙總語句無法合并的情況發生。

在視窗資料表值函式和彙總語句之間，包含對window_start、window_end和window_time欄位的過濾或計算。例如：

-- 包含對window_start的過濾
> SELECT window_start, window_end, item, SUM(price) AS total_price
    FROM
    (SELECT item, price, window_start, window_end FROM
    TABLE(
    SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
    where window_start >= TIMESTAMP '2020-04-15 08:06:00.000')
    GROUP BY item, window_start, window_end;
  
-- 包含對window_start的計算
> SELECT window_start, window_end, item, SUM(price) AS total_price
    FROM
    (SELECT item, price, window_start + (INTERVAL '1' SECOND) as window_start, window_end FROM
    TABLE(
    SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
    GROUP BY item, window_start, window_end;

-- 包含對window_start的計算
> SELECT window_start, window_end, item, SUM(price) AS total_price
    FROM
    (SELECT item, price, CAST(window_start as varchar) as window_start, window_end FROM
    TABLE(
    SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
    GROUP BY item, window_start, window_end;

視窗資料表值函式和UDTF同時使用。例如：

> SELECT window_start, window_end, category, SUM(price) AS total_price
    FROM
    (SELECT category, price, window_start, window_end FROM
    TABLE(
    SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)),
    LATERAL TABLE(category_udtf(item)) as T(category))
    GROUP BY category, window_start, window_end;

彙總語句的GROUP KEY中未同時包含window_start和window_end。例如：

 SELECT window_start, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start;

彙總函式使用python UDAF。

彙總函式使用GROUPING SETS、CUBE和ROLLUP文法，導致window_start和window_end不在同一組GROUP KEY中。例如：

> SELECT item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY GROUPING SETS((item), (window_start), (window_end));
  
> SELECT item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY CUBE (item, window_start, window_end);
  
> SELECT item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY ROLLUP (item, window_start, window_end);

彙總語句中，彙總函式使用視窗列window_start、window_end和window_time進行計算。例如：

> SELECT window_start, window_end, item, SUM(price) AS total_price, max(window_end) AS max_end
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

新老文法對更新流的支援情況

視窗函數	老文法（GroupWindowAggregation運算元）	新文法（WindowAggregate運算元）
視窗函數	VVR、社區Flink	VVR	社區Flink
TUMBLE	支援	支援	不支援
HOP	支援	支援	不支援
SESSION	支援	支援說明 VVR和社區Flink關於Session視窗區別請參見Queries語句。	Flink 1.19支援
CUMULATE	N/A	支援說明 Realtime Compute引擎VVR 8.0.6及以上版本支援。	不支援

在對更新流的支援上，老文法視窗彙總（GroupWindowAggregation運算元）支援更新流（VVR和社區Flink保持一致），新文法（WindowAggregate運算元）社區Flink（1.16~1.18）不支援更新流，而VVR實現了新老文法的內部融合，可以自動根據輸入資料流的情況選擇支援的運算元，實現社區Flink新文法中不支援更新流的TUMBLE、HOP視窗彙總對更新流的支援。

視窗彙總 Local-Global 最佳化

視窗彙總支援 Local-Global 兩階段彙總最佳化。啟用後，最佳化器將單階段視窗彙總拆分為：

Local Aggregate：在資料 Shuffle 之前做局部預彙總，減少網路傳輸資料量。
Global Aggregate：在 Shuffle 之後做最終彙總，輸出結果。

該最佳化需同時滿足以下 6 個條件方可生效。

條件 1：彙總階段策略允許兩階段

table.optimizer.agg-phase-strategy 設定為 AUTO（預設）或 TWO_PHASE。若設為 ONE_PHASE，則強制禁用兩階段最佳化。

條件 2：視窗基於事件時間

視窗必須基於事件時間（Event Time / Rowtime），不支援處理時間視窗。

條件 3：視窗類別型非 Session

支援 TUMBLE、HOP、CUMULATE 視窗類別型。Session 視窗不支援兩階段最佳化。

條件 4：彙總函式支援 Partial Merge

所有彙總函式必須支援合併作業。內建的 SUM、COUNT、MIN、MAX、AVG 均支援。自訂 UDAF 需實現 merge() 方法。

條件 5：輸入資料流為 Insert-Only 且視窗可轉換為 TVF 形式

需同時滿足以下條件：

輸入資料流為 Insert-Only。
table.exec.emit.early-fire.enabled 為 false（預設）。
table.exec.emit.late-fire.enabled 為 false（預設）。
HOP 視窗必須對齊（視窗大小可被滑動步長整除）。

條件 6：資料分布未滿足分區要求

輸入資料的分布尚未滿足彙總所需的分區要求。若資料已按分區鍵分布，最佳化器判斷無需額外預彙總，不會產生 Local Aggregate 節點。

參數速查

參數	類型	預設值	生效條件
`table.optimizer.agg-phase-strategy`	Enum	AUTO	不能為 ONE_PHASE
`table.exec.emit.early-fire.enabled`	Boolean	false	必須為 false
`table.exec.emit.late-fire.enabled`	Boolean	false	必須為 false