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

SESSION ウィンドウ TVF: VVR 11 vs VVR 8

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

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

次の表は、VVR 11.x と VVR 8.x のセッションウィンドウ TVF を比較しています。

次の SQL の例は同等であり、どちらも item を SESSION ウィンドウ関数のパーティションフィールドとして使用しています。

-- テーブルには時間属性が必要です (このテーブルでは `bidtime` など)
> 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;

SQL コードの比較:

集計クエリにおけるウィンドウ TVF の制限

SESSION ウィンドウ TVF を例として、集約クエリでウィンドウ TVF を使用できないユースケースについて説明します。

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

• ウィンドウ 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;

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

• Python ユーザー定義集計関数（UDAF）が使用されます。

• GROUP BY 句で 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);

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