Realtime Compute for Apache Flink:Flink SQL の最適化

JobManager リソースのスケーリング

並列デプロイメントの場合、設定 タブの リソース セクションで JobManager のリソースを増やします。例：

• Job Manager CPU：4

• Job Manager Memory：8 GiB

TaskManager リソースのスケーリング

複雑なトポロジーの場合、設定 タブの リソース セクションで TaskManager のリソースを増やします。例：

• Task Manager CPU：2

• Task Manager Memory：4 GiB

miniBatch の有効化

MiniBatch は、受信レコードをバッファリングして一括で処理することで、バッチごとの状態アクセスを削減します。これにより、わずかに高いレイテンシーと引き換えにスループットが向上します。miniBatch 機能は、指定された間隔でソースに挿入されるイベントメッセージに基づいてマイクロバッチ処理をトリガーします。

利用シーン：超低レイテンシーが要求されないデータ集計シナリオ。

有効化の方法：設定 タブの パラメーター セクションにある その他の設定 フィールドに以下を追加します。詳細については、「カスタムデプロイメントパラメーターの設定」をご参照ください。

table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5s

LocalGlobal の有効化

LocalGlobal は、MapReduce の combine と reduce に似て、集計を 2 つのフェーズに分割します。つまり、各上流ノードでのローカル事前集計と、その後のグローバル集計です。ローカル集計はホットスポットデータを事前にフィルタリングし、グローバルフェーズに到達するデータ量を削減します。

利用シーン：SUM、COUNT、MAX、MIN、AVG などの一般的な集計関数で、データホットスポットの問題がある場合。

前提条件：

• MiniBatch が有効になっている必要があります。LocalGlobal は、ミニバッチの間隔を使用して、ローカル集計の前に蓄積するレコード数を決定します。

• 集計関数は merge メソッド (AggregateFunction) を実装している必要があります。

ステータス：miniBatch がアクティブな場合、デフォルトで有効になります。

検証：最終的なトポロジーで GlobalGroupAggregate または LocalGroupAggregate ノードを確認します。

COUNT DISTINCT に対する PartialFinal の有効化

ローカル集計では重複する distinct キーを削除できないため、LocalGlobal は COUNT DISTINCT に対しては効果が低くなります。PartialFinal は、集計を 2 つのレイヤーに分割することでこの問題に対処します。最初のレイヤーは distinct キーのハッシュによってデータを分散させ、2 番目のレイヤーが最終的な集計を実行します。

利用シーン：データ量が大きく、集計パフォーマンスがボトルネックとなっている COUNT DISTINCT クエリ。

有効化の方法：その他の設定 フィールドに以下を追加します。詳細については、「カスタムデプロイメントパラメーターの設定」をご参照ください。

table.optimizer.distinct-agg.split.enabled: true

制限事項：

• PartialFinal は、ユーザー定義集約関数 (UDAF) を含む SQL では使用できません。

• データ量が大きい場合にのみ有効にしてください。PartialFinal は追加のネットワークシャッフルを導入します。

検証：最終的なトポロジーで、1 レイヤー集計が 2 レイヤー集計に変わったかどうかを確認します。

CASE WHEN の代わりに AGG WITH FILTER を使用

異なる条件下で同じフィールドに対して COUNT DISTINCT を計算する場合、CASE WHEN の代わりに FILTER 句を使用します。オプティマイザーは、同じ distinct キーに対する異なるフィルター引数を認識し、単一の共有ステートインスタンスを使用するため、状態アクセスとサイズが削減されます。パフォーマンステストでは、CASE WHEN に比べて 2 倍の改善が見られました。

変更前 (非効率)：

COUNT(DISTINCT visitor_id) AS uv_total,
COUNT(DISTINCT CASE WHEN is_wireless = 'y' THEN visitor_id ELSE NULL END) AS uv_wireless

変更後 (2 倍のパフォーマンス)：

COUNT(DISTINCT visitor_id) AS uv_total,
COUNT(DISTINCT visitor_id) FILTER (WHERE is_wireless = 'y') AS uv_wireless

デルタ集計 (ステートレス集計) の使用

デルタ集計は、ストリーミングの非ウィンドウグループ集計に対してステートレスな集計を提供します。Flink の状態でグループごとに集計アキュムレータを継続的に維持する通常のグループ集計とは異なり、デルタ集計はデータ変更に応答します。グループ化キーを使用してグループの現在の詳細レコードを非同期でルックアップし、それらのレコードから集計結果を再計算します。

その結果、Flink 側の集計状態の量は、グループの履歴数に依存しなくなります。デプロイメントは、グループ数が増えるにつれて集計状態を継続的に拡張する必要がありません。これにより、チェックポイントに永続化されるデータ量が削減され、回復時間が短縮され、大規模なグループ集計デプロイメントのリソース使用量が安定します。

SET 'table.exec.async-lookup.buffer-capacity' = '100';
SET 'table.exec.async-lookup.timeout' = '3 min';

SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '10s';
SET 'table.exec.mini-batch.size' = '1000';

SET 'table.exec.delta-agg.cache-enabled' = 'true';
SET 'table.exec.delta-agg.cache-size' = '2000';

+U(..., 120) → 条件に一致し、結果テーブルに書き込まれる
+U(..., 80)  → 条件に一致せず、フィルターで除外される

SET 'table.optimizer.delta-agg.ignore-non-unique-key-filter' = 'true';

2ストリーム結合におけるキーバリュー分離

VVR 6.0.1 以降では、エンジンは 2 ストリーム JOIN 演算子に対してキーバリュー分離を有効にするかどうかを自動的に推論し、典型的な結合パフォーマンスを 40% 以上向上させます。

この動作を制御するには、table.exec.join.kv-separate パラメーターを設定します。

結合に対する miniBatch の有効化

デフォルトでは、通常の結合演算子はレコードを 1 つずつ処理します。つまり、結合キーで相手側の状態をルックアップし、状態を更新し、結果を出力します。このレコードごとのパターンは、特に RocksDB を使用している場合に状態バックエンドのオーバーヘッドを増加させ、カスケード結合では深刻なレコード増幅を引き起こす可能性があります。

結合のための MiniBatch は、2 つのコアな最適化でこれに対処します。

1. レコードの折りたたみ：バッファー内のレコードを折りたたむことで、結合プロセス前のデータ量を削減します。

2. 出力の抑制：バッチ処理中に冗長な中間結果を抑制します。

MiniBatch は、最大許容レイテンシーに達したとき、バッチサイズ制限に達したとき、またはチェックポイントが発生したときに処理をトリガーします。

VVR バージョン：8.0.4 以降。

有効化の方法：その他の設定 フィールドに以下を追加します。詳細については、「カスタムデプロイメントパラメーターの設定」をご参照ください。

table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5s
table.exec.stream.join.mini-batch-enabled: true

最適なケース：メッセージ増幅を伴うカスケード外部結合。上流の演算子はメッセージの折りたたみまたはマージを使用して、下流の演算子での増幅を抑制します。例：

SELECT a.id AS a_id, a.a_content, B.id AS b_id, B.b_content
FROM a LEFT JOIN
  (SELECT * FROM b
      LEFT JOIN c ON b.prd_id = c.id) B
ON a.id = B.id

TopN アルゴリズム

使用されるアルゴリズムは、入力データストリームが静的か動的かによって異なります。

アルゴリズム名はトポロジーノード名に表示されます。

RetractRank から UpdateFastRank への切り替え

UpdateFastRank には、以下のすべての条件が必要です。

1. 入力ストリームに DELETE または UPDATE_BEFORE メッセージが含まれていないこと。EXPLAIN で確認します。

      EXPLAIN CHANGELOG_MODE <query_statement_or_insert_statement_or_statement_set>

2. 入力ストリームにプライマリキー情報 (例：GROUP BY 句で使用される列) が含まれていること。

3. ORDER BY フィールドが、ソート順とは逆の順序で単調に更新されること。例：ORDER BY COUNT DESC、ORDER BY COUNT_DISTINCT DESC、または ORDER BY SUM(正の値) DESC。

例：ORDER BY SUM DESC の場合、正の値にフィルタリングして単調性を保証します。

INSERT INTO print_test
SELECT
  cate_id,
  seller_id,
  stat_date,
  pay_ord_amt
FROM (
    SELECT
      *,
      ROW_NUMBER() OVER (
        -- PARTITION BY 列はサブクエリの GROUP BY に現れる必要があります。
        -- 状態が期限切れになったときの順序の乱れを防ぐために時間フィールドを含めます。
        PARTITION BY cate_id, stat_date
        ORDER BY pay_ord_amt DESC
      ) AS rownum
    FROM (
        SELECT
          cate_id,
          seller_id,
          stat_date,
          -- 正の値のみが含まれるため、SUM は単調増加します。
          SUM(total_fee) FILTER (WHERE total_fee >= 0) AS pay_ord_amt
        FROM random_test
        WHERE total_fee >= 0
        GROUP BY seller_id, stat_date, cate_id
      ) a
  )
WHERE rownum <= 100;

出力ボリュームの削減

最終的な SELECT 出力に rownum を含めないでください。代わりにフロントエンドで結果をソートすることで、sink の書き込み量を削減します。詳細については、「Top-N」をご参照ください。

TopN キャッシュサイズの増加

TopN は、ディスクの読み取りを減らすために状態キャッシュを維持します。キャッシュヒット率は次の数式で計算します。

キャッシュヒット率 = キャッシュサイズ * 並列度 / top_n / パーティションキー数

例：Top100、デフォルトのキャッシュサイズ 10,000 レコード、並列度 50、パーティションキー数 100,000 の場合：

10,000 * 50 / 100 / 100,000 = 5% のヒット率

5% のヒット率は、ほとんどの読み取りがディスクに向かうことを意味し、state seek メトリックが不安定になり、パフォーマンスが低下します。キャッシュサイズを増やします。

table.exec.rank.topn-cache-size: 200000

200,000 のキャッシュエントリの場合：

200,000 * 50 / 100 / 100,000 = 100% のヒット率

パーティションキー数が多い場合は、TopN キャッシュサイズとヒープメモリも増やしてください。詳細については、「デプロイメントの設定」をご参照ください。

PARTITION BY への時間フィールドの追加

day などの時間フィールドを PARTITION BY 句に追加します。これがないと、TTL により状態データが期限切れになったときに TopN の結果が乱れます。

構文

重複排除は TopN の特殊なケースです。ROW_NUMBER() と OVER 句を使用してパーティションごとに行番号を割り当て、rownum = 1 のみ保持します。

SELECT *
FROM (
   SELECT *,
    ROW_NUMBER() OVER (PARTITION BY col1[, col2..]
     ORDER BY timeAttributeCol [ASC|DESC]) AS rownum
   FROM table_name)
WHERE rownum = 1

時間属性の動作：

• proctime (処理時間)：重複排除はレコードが処理された時間に基づきます。結果は実行ごとに異なる場合があります。

• rowtime (イベント時間)：重複排除はレコードが生成された時間に基づきます。結果は決定的です。

FirstRow の保持

重複排除キーごとに最初のレコードを保持します。状態はプライマリキーデータのみを保存するため、状態アクセスは非常に効率的です。

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b ORDER BY proctime) AS rowNum
  FROM T
)
WHERE rowNum = 1

テーブル T を列 b で重複排除し、処理時間で最も早いレコードを保持します。proctime 属性を宣言する代わりに、PROCTIME() 関数を呼び出します。

LastRow の保持

重複排除キーごとに最新のレコードを保持します。これは LAST_VALUE 関数よりもわずかにパフォーマンスが優れています。

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, d ORDER BY rowtime DESC) AS rowNum
  FROM T
)
WHERE rowNum = 1

テーブル T を列 b と d で重複排除し、イベント時間で最新のレコードを保持します。

UDF よりもビルトイン関数を優先

ビルトイン関数は、シリアル化、デシリアル化、およびバイトレベルの操作に最適化されています。可能な限り、UDF をビルトインの同等機能に置き換えてください。

KEYVALUE における 1 文字デリミタの使用

KEYVALUE(content, keyValueSplit, keySplit, keyName) 関数は、keyValueSplit と keySplit が : や , などの 1 文字である場合、約 30% 高速に実行されます。1 文字のデリミタを使用すると、エンジンは入力全体を解析することなく、バイナリデータ内で直接ターゲットキーを検索します。

LIKE 演算子のパターン

アンダースコアのエスケープ：アンダースコア (_) は SQL における 1 文字のワイルドカードです。リテラルのアンダースコアに一致させるには、エスケープ文字を使用します。

LIKE '%seller/_id%' ESCAPE '/'

エスケープしない場合、LIKE '%seller_id%' は seller#id、sellerxid、seller1id、およびその他の意図しない文字列にも一致します。

正規表現の回避

正規表現は算術演算よりも 100 倍遅くなる可能性があり、エッジケースでは無限ループに陥り、デプロイメントをブロックすることがあります。可能な場合は、代わりに LIKE を使用してください。

正規表現が必要なケースについては、「REGEXP」および「REGEXP_REPLACE」をご参照ください。

構文

構文は Apache Calcite SQL に従います。

SELECT /*+ hint [, hint ] */ ...

-- ここで:
-- hint:       hintName(hintOption [, hintOption]*)
-- hintOption:  simpleIdentifier | numericLiteral | stringLiteral

結合ヒント

クエリヒントは、現在の クエリブロック 内の実行計画を変更します。Flink は現在、ディメンションテーブル結合 と 通常結合 の両方に対して、結合ヒントのみをサポートしています。