Realtime Compute for Apache Flink:ウィンドウ集計

SESSION ウィンドウ TVF 構文：VVR 11.x と VVR 8.x の比較

SESSION ウィンドウ TVF 構文は VVR バージョンによって異なります。完全な機能の構文を使用するには、VVR 11.1 以降にアップグレードしてください。

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

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

-- Bid テーブルスキーマ (以下のすべての例で使用)
> 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：パーティションフィールドを SESSION() で明示的に宣言
> 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：パーティションフィールドを GROUP BY から推測
> 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;

ウィンドウ TVF 集計が非 TVF モードにフォールバックする場合

クエリにウィンドウ TVF が含まれているが、TVF と集計をマージする条件を満たしていない場合、システムは非 TVF の実行計画にフォールバックします。

次のいずれかの条件を満たす場合、ウィンドウ TVF と集計ステートメントをマージできません。

1. ウィンドウ時間フィールドに対するフィルタリングまたは計算： 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;

2. ウィンドウ TVF がユーザー定義テーブル関数 (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;

3. GROUP BY 句に 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;

4. Python のユーザー定義集計関数 (UDAF) が使用されている場合。

5. GROUPING SETS、CUBE、または ROLLUP が window_start または window_end で個別にグループ化している場合。

   > 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);

6. 集計関数が 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;

条件1：集計段階の戦略が 2 段階を許可

table.optimizer.agg-phase-strategy が AUTO (デフォルト) または TWO_PHASE に設定されている必要があります。ONE_PHASE に設定すると、2 段階最適化が無効になります。

条件2：ウィンドウがイベント時間を使用

ウィンドウはイベント時間 (rowtime) を使用する必要があります。処理時間ウィンドウはサポートされていません。

条件3：ウィンドウタイプが SESSION でない

TUMBLE、HOP、CUMULATE ウィンドウタイプがサポートされています。SESSION ウィンドウは 2 段階最適化をサポートしていません。

条件4：すべての集計関数が部分マージをサポート

すべての集計関数がマージ操作をサポートしている必要があります。SUM、COUNT、MIN、MAX、AVG などの組み込み関数がサポートされています。カスタム UDAF は merge() メソッドを実装する必要があります。

条件5：入力ストリームが挿入のみで、ウィンドウが TVF 形式に変換可能

次の条件をすべて満たす必要があります。

• 入力ストリームが挿入のみであること。

• table.exec.emit.early-fire.enabled が false (デフォルト) に設定されていること。

• table.exec.emit.late-fire.enabled が false (デフォルト) に設定されていること。

• HOP ウィンドウの場合、ウィンドウが整列していること (ウィンドウサイズがスライド間隔で割り切れる)。

条件6：データ分散がパーティション要件を事前に満たしていない

入力データの分散が集計のパーティション要件をまだ満たしていない必要があります。データが既にパーティションキーで分散されている場合、オプティマイザは追加の事前集計が不要であると判断し、Local Aggregate ノードを生成しません。

パラメーターリファレンス