Realtime Compute for Apache Flink:ウィンドウ Top-N 関数

状態の互換性を破壊する変更の理由

ウィンドウ Top-N クエリを変更すると、Flink のクエリ オプテマイザーが異なる実行計画を生成することがあります。これにより、演算子のトポロジーや中間演算子の状態スキーマが変更される可能性があります。新しい計画が既存のセーブポイントにマップされない場合、タスクは以前の状態から再開できません。

ウィンドウ属性、GROUP BY フィールド、パーティションキー、ORDER BY フィールド、N の値、または集約フィールドを変更する操作は、再計画をトリガーし、状態の非互換性を引き起こします。

ウィンドウ Top-N の構文構造

各変更がどの句を対象としているかを理解することで、以下に示す互換性ルールを適切に適用できます。ウィンドウ Top-N クエリの構造は次のとおりです。

-- 外側のクエリ：上位ランクの行を選択
SELECT [column_list]
FROM (
  -- 中間レイヤー：ROW_NUMBER() を使用して行番号を割り当てる
  SELECT [column_list],
    ROW_NUMBER() OVER (
      PARTITION BY window_start, window_end [, partition_key...]
      ORDER BY col [ASC|DESC] [, col [ASC|DESC]...]
    ) AS rk
  FROM (
    -- 内側のクエリ：ウィンドウ TVF を使用したウィンドウ集計
    SELECT [agg_fields], window_start, window_end
    FROM TABLE(TUMBLE(TABLE source_table, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY [group_keys], window_start, window_end
  )
)
WHERE rk < N;

以下の互換性ルールは、この構造内の特定の句に対応しています。

互換性のある変更

以下の変更は、既存の状態データと完全に互換性があります。タスクの状態をリセットすることなく実施できます。

-- 元のクエリ：window_start のみを選択
SELECT a, b, c, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, window_start, window_end
  )
) WHERE rk < 3;

-- 互換性あり：出力に window_end を追加
SELECT a, b, c, window_start, window_end FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, window_start, window_end
  )
) WHERE rk < 3;

-- 元のクエリ：出力に rk は含まれない
SELECT a, b, c, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, window_start, window_end
  )
) WHERE rk < 3;

-- 互換性あり：出力に rk を含める
SELECT a, b, c, window_start, rk FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, window_start, window_end
  )
) WHERE rk < 3;

-- 元のクエリ：PARTITION BY a, b, window_start, window_end
SELECT a, b, c, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a, b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, b, window_start, window_end
  )
) WHERE rk < 3;

-- 互換性あり：PARTITION BY b, a, window_start, window_end に順序を変更
SELECT a, b, c, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, a, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, b, window_start, window_end
  )
) WHERE rk < 3;

互換性のない変更

以下の変更は、既存の状態データと互換性がありません。これらの変更を実施した後は、タスクを再起動する前に状態をリセットする必要があります。

-- 元のクエリ
SELECT a, b, c, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, window_start, window_end
  )
) WHERE rk < 3;

-- 非互換：min(d) as d を追加し、Top-N 入力を変更
SELECT a, b, c, d, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, min(d) AS d, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, window_start, window_end
  )
) WHERE rk < 3;

-- 元のクエリ
SELECT a, b, c, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, b, window_start, window_end
  )
) WHERE rk < 3;

-- 非互換：ウィンドウ TVF フィールドの順序を変更（window_end を window_start の前に）
-- これにより、ウィンドウベースのランキング結果が変化します。
SELECT a, b, c, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY a, b, window_end, window_start
  )
) WHERE rk < 3;

-- 非互換：ウィンドウ以外のフィールドの順序を変更（b を a の前に）
SELECT a, b, c, window_start FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, window_start, window_end ORDER BY c) AS rk
  FROM (
    SELECT a, sum(b) AS b, max(c) AS c, window_start, window_end
    FROM TABLE(tumble(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
    GROUP BY b, a, window_start, window_end
  )
) WHERE rk < 3;

次のステップ

• 完全な非互換性を引き起こす変更

• 互換性のない変更