OVER ウィンドウ - Realtime Compute for Apache Flink - Alibaba Cloud ドキュメントセンター

OVER ウィンドウは、従来のデータベースで使用される標準ウィンドウです。Over 集計は、ウィンドウ集計とは異なります。OVER ウィンドウを使用するストリーミングデータでは、各要素が OVER ウィンドウに対応します。OVER ウィンドウは、要素の実際の行または実際の値（タイムスタンプ値）に基づいて決定できます。ストリームの要素は、複数のウィンドウに分散されます。

OVER ウィンドウを適用するストリームでは、各要素が OVER ウィンドウに対応し、データ計算を 1 回トリガーします。計算をトリガーする各要素によって決定される行は、要素が配置されているウィンドウの最後の行です。Realtime Compute の基盤となる実装では、OVER ウィンドウデータは一元管理されます。データのコピーは 1 つだけ保存されます。論理的には、各要素に対して OVER ウィンドウが作成されます。Realtime Compute for Apache Flink は、各 OVER ウィンドウのデータを計算し、計算の完了後に不要になったデータを削除します。詳細については、「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) は同じである必要があります。
AS で指定されたエイリアスは、外部 SQL ステートメントを使用してクエリできます。

ウィンドウタイプ

Flink SQL では、OVER ウィンドウは標準 SQL 構文に準拠して定義されます。従来の OVER ウィンドウは、きめ細かいウィンドウタイプに分類されません。OVER ウィンドウは、計算対象行の決定方法に基づいて、次の 2 つのタイプに分類されます。

ROWS OVER ウィンドウ: 要素の各行は新しい計算対象行として扱われます。行ごとに新しいウィンドウが生成されます。
RANGE OVER ウィンドウ: タイムスタンプ値が同じ要素のすべての行は、1 つの計算対象行として扱われ、同じウィンドウに割り当てられます。

属性

直交属性	説明	proctime	eventtime
ROWS OVER ウィンドウ	ウィンドウは、要素の実際の行に基づいて決定されます。	サポート対象	サポート対象
RANGE OVER ウィンドウ	ウィンドウは、要素のタイムスタンプ値に基づいて決定されます。	サポート対象	サポート対象

ROWS OVER ウィンドウ

説明
ROWS OVER ウィンドウの場合、要素ごとにウィンドウが生成されます。
構文
```
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: 現在の行の前にある行数を指定します。

例

この例では、境界付き ROWS OVER ウィンドウについて説明します。この例では、販売中の商品テーブルに、商品 ID、商品タイプ、発売時刻、価格が含まれています。現在の商品が販売される前の、現在の商品に類似した 3 つの商品のうち、最も高い価格を計算します。

テストデータ

itemid（VARCHAR）	itemtype（VARCHAR）	eventtime（VARCHAR）	price（DOUBLE）
ITEM001	電子機器	`2024-11-11 10:01:00`	20
ITEM002	電子機器	`2024-11-11 10:02:00`	50
ITEM003	電子機器	`2024-11-11 10:03:00`	30
ITEM004	電子機器	`2024-11-11 10:03:00`	60
ITEM005	電子機器	`2024-11-11 10:05:00`	40
ITEM006	電子機器	`2024-11-11 10:06:00`	20
ITEM007	電子機器	`2024-11-11 10:07:00`	70
ITEM008	衣類	`2024-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 '2' SECOND  -- Rowtime のウォーターマークを定義します。
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<brokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

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	電子機器	`2024-11-11 10:01:00`	20	20
ITEM002	電子機器	`2024-11-11 10:02:00`	50	50
ITEM003	電子機器	`2024-11-11 10:03:00`	30	50
ITEM004	電子機器	`2024-11-11 10:03:00`	60	60
ITEM005	電子機器	`2024-11-11 10:05:00`	40	60
ITEM006	電子機器	`2024-11-11 10:06:00`	20	60
ITEM007	電子機器	`2024-11-11 10:07:00`	70	70
ITEM008	衣類	`2024-11-11 10:08:00`	20	20

RANGE OVER ウィンドウ

説明
RANGE OVER ウィンドウの場合、タイムスタンプ値が同じすべての要素が同じウィンドウに割り当てられます。
構文
```
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: 現在の行の時刻と、それが追跡できる要素行の時刻の間の時間間隔を指定します。

例

この例では、境界付き RANGE OVER ウィンドウについて説明します。この例では、販売中の商品テーブルに、商品 ID、商品タイプ、発売時刻、価格が含まれています。現在の商品よりも 2 分前に販売された類似商品の中で最も高い価格を計算します。

テストデータ

itemid（VARCHAR）	itemtype（VARCHAR）	eventtime（VARCHAR）	price（DOUBLE）
ITEM001	電子機器	`2024-11-11 10:01:00`	20
ITEM002	電子機器	`2024-11-11 10:02:00`	50
ITEM003	電子機器	`2024-11-11 10:03:00`	30
ITEM004	電子機器	`2024-11-11 10:03:00`	60
ITEM005	電子機器	`2024-11-11 10:05:00`	40
ITEM006	電子機器	`2024-11-11 10:06:00`	20
ITEM007	電子機器	`2024-11-11 10:07:00`	70
ITEM008	衣類	`2024-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 '2' SECOND  -- Rowtime のウォーターマークを定義します。
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<brokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

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	電子機器	`2024-11-11 10:01:00`	20	20
ITEM002	電子機器	`2024-11-11 10:02:00`	50	50
ITEM003	電子機器	`2024-11-11 10:03:00`	30	50
ITEM004	電子機器	`2024-11-11 10:03:00`	60	60
ITEM005	電子機器	`2024-11-11 10:05:00`	40	60
ITEM006	電子機器	`2024-11-11 10:06:00`	20	40
ITEM007	電子機器	`2024-11-11 10:07:00`	70	70
ITEM008	衣類	`2024-11-11 10:08:00`	20	20