Realtime Compute for Apache Flink:ジョブのパフォーマンスに関する問題

オペレーターノードの分割

pipeline.operator-chaining: 'false'

グループ集計の最適化

• MiniBatch の有効化 (スループットの向上)

  MiniBatch は、処理をトリガーする前に流入データをバッファリングします。これにより、State へのアクセス頻度が低下し、スループットが向上し、出力ボリュームが減少します。

  MiniBatch は、ソースで指定した間隔で挿入されるイベントメッセージに基づいて、マイクロバッチ処理をトリガーします。

  • シナリオ

    マイクロバッチ処理は、レイテンシーがわずかに増える代わりに、スループットを大幅に向上させます。超低レイテンシーが必要な場合は有効化しないでください。ほとんどの集計シナリオでは、MiniBatch を有効化することでシステムパフォーマンスが大きく向上します。

  • 有効化方法

    MiniBatch はデフォルトで無効になっています。有効にするには、対象のジョブの [デプロイ詳細] タブに移動し、[その他の設定] の下の [ランタイムパラメーター設定] セクションで、次のコードを追加します。

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

    パラメーターの説明を次の表に示します。

    パラメーター	説明
    table.exec.mini-batch.enabled	ミニバッチを有効化するかどうかを指定します。
    table.exec.mini-batch.allow-latency	バッチ出力間の時間間隔です。

• LocalGlobal の有効化 (一般的なデータホットスポット問題の解消)

  LocalGlobal メカニズムは、LocalAgg を使用してスキューしたデータを事前にローカル集計し、GlobalAgg にかかるホットスポットの負荷を軽減して、全体のパフォーマンスを向上させます。

  LocalGlobal は、単一の集計を Local と Global の 2 段階に分割します。これは MapReduce の Combine フェーズと Reduce フェーズに似ています。第 1 段階では、上流ノードがデータをローカルにバッファリングして集計 (localAgg) し、増分アキュムレーターを出力します。第 2 段階では、これらのアキュムレーターをマージ (Merge) して最終結果 (GlobalAgg) を生成します。

  • シナリオ

    標準的な集計 (SUM、COUNT、MAX、MIN、AVG など) のパフォーマンスを向上させ、これらのシナリオにおけるデータホットスポットの問題を解消します。

  • 制限事項

    LocalGlobal はデフォルトで有効ですが、以下の制限事項があります：

    • MiniBatch を有効化する必要があります。

    • AggregateFunction は Merge を実装する必要があります。

  • 有効化の確認

    生成されたトポロジーに、GlobalGroupAggregate または LocalGroupAggregate という名前のノードが含まれていることを確認します。

• PartialFinal の有効化 (COUNT DISTINCT ホットスポット問題の解消)

  COUNT DISTINCT のホットスポットに対処するには、従来はクエリを手動で 2 段階集計 (modulo ベースのシャッフル層を追加) に書き換える必要がありました。Realtime Compute for Apache Flink は、PartialFinal 最適化により COUNT DISTINCT のシャッフルを自動的に実行するため、手動での書き換えは不要です。

  LocalGlobal は標準的な集計では効果的ですが、COUNT DISTINCT では効果が限定的です。ローカル集計中の distinct キーの重複排除率が低いままであるため、グローバルノードでホットスポットが継続します。

  • シナリオ

    COUNT DISTINCT が集計ノードのパフォーマンス要件を満たさない場合に使用します。

    重要

    • UDAF を含む Flink SQL では PartialFinal 最適化を使用しないでください。

    • データ量が少ない場合は PartialFinal を避けてください。不要なネットワークシャッフルが発生し、リソースを無駄にします。

  • 有効化方法

    この機能はデフォルトで無効になっています。有効にするには、対象ジョブの [デプロイ詳細] タブの [実行パラメーター設定] セクションで、[その他の設定] フィールドに次のコードを入力します。

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

  • 有効化の確認

    生成されたトポロジーの集計ステージが、単一ステージから 2 段階に変わっていることを確認します。

• AGG WITH CASE WHEN から AGG WITH FILTER 構文への書き換え (複数の COUNT DISTINCT シナリオにおけるパフォーマンス向上)

  ジョブで、総 UV、モバイルクライアント UV、PC UV など、複数のディメンションにまたがる UV を計算する場合は、CASE WHEN ではなく標準の AGG WITH FILTER 構文を使用することを推奨します。Realtime Compute for Apache Flink の SQL オプティマイザーは Filter パラメーターを認識するため、同一フィールドに対する複数の COUNT DISTINCT で State を共有でき、State I/O を削減できます。パフォーマンステストでは、この書き換えによりパフォーマンスが 2 倍になることが示されています。

  • シナリオ

    同一フィールドに対して、条件の異なる複数の COUNT DISTINCT 結果を計算する場合に、大幅なパフォーマンス向上が得られます。

  • 最適化前の構文

    COUNT(distinct visitor_id) as UV1, COUNT(distinct case when is_wireless='y' then visitor_id else null end) as UV2

  • 最適化後の構文

    COUNT(distinct visitor_id) as UV1, COUNT(distinct visitor_id) filter (where is_wireless='y') as UV2

TopN の最適化

• TopN アルゴリズム

  TopN の入力が追記専用ストリーム (例：SLS から) の場合、使用できるアルゴリズムは AppendRank のみです。入力が更新ストリーム (例：AGG または JOIN の後) の場合、使用できるアルゴリズムは 2 つあり、パフォーマンスの高い順に UpdateFastRank、RetractRank です。アルゴリズム名はトポロジーのノードラベルに表示されます。

  • AppendRank：追記専用ストリームでのみサポートされます。

  • UpdateFastRank：更新ストリームに最適です。

  • RetractRank：更新ストリーム向けのフォールバックアルゴリズムです。パフォーマンスは低くなります。場合によっては UpdateFastRank に最適化できます。

  RetractRank を UpdateFastRank に最適化するには、次の 3 つの条件を満たす必要があります：

  • 入力ストリームが更新ストリームである必要があります。

  • 入力ストリームに Primary Key 情報が含まれている必要があります (例：GROUP BY 集計の後など)。

  • ソートフィールドが、ソート順と逆方向に単調更新される必要があります (例：ORDER BY COUNT、COUNT_DISTINCT、または SUM (正の値) DESC)。

  ORDER BY SUM DESC で UpdateFastRank が使用されるようにするには、[total_fee] が正であることを保証するフィルター条件を追加します。

  insert
    into print_test
  SELECT
    cate_id,
    seller_id,
    stat_date,
    pay_ord_amt  -- シンクテーブルの出力を減らすため、rownum を省略します。
  FROM (
      SELECT
        *,
        ROW_NUMBER () OVER (
          PARTITION BY cate_id,
          stat_date  -- State TTL によるデータ破損を防ぐため、時間フィールドを含めます。
          ORDER
            BY pay_ord_amt DESC
        ) as rownum  -- 上流の sum 結果でソートします。
      FROM (
          SELECT
            cate_id,
            seller_id,
            stat_date,
            -- 重要：すべての SUM 入力を正として宣言し、単調増加を保証します。
            -- これにより、TopN は最適化アルゴリズムを使用でき、上位 100 レコードのみを取得できます。
            sum (total_fee) filter (
              where
                total_fee >= 0
            ) as pay_ord_amt
          FROM
            random_test
          WHERE
            total_fee >= 0
          GROUP
            BY cate_name,
            seller_id,
            stat_date,
            cate_id
        ) a
      ) WHERE
        rownum <= 100;

• TopN の最適化方法

  • ランキング出力がない場合の最適化

    TopN の出力で rownum 値を表示する必要がない場合は、rownum を省略し、フロントエンドで 1 回だけソートします。これにより、シンクテーブルの出力ボリュームが大幅に減少します。詳細については、「Top-N」をご参照ください。

  • TopN キャッシュサイズの引き上げ

    TopN は、State へのアクセス効率を向上させるために State Cache レイヤーを使用します。キャッシュヒット率は次の式で計算されます。

    cache_hit = cache_size * parallelism / top_n / partition_key_num

    たとえば、Top100、キャッシュサイズ 10,000、並列度 50、パーティションキー 100,000 の場合、ヒット率は 10000 * 50 / 100 / 100000 = 5% にすぎません。ヒット率が低いと、ほとんどのリクエストがディスクベースの State に到達し、state seek メトリクスにスパイクが発生して、パフォーマンスが大幅に低下します。

    パーティションキーのカーディナリティが非常に高い場合は、TopN キャッシュサイズとヒープメモリを適宜増やしてください。詳細については、「ジョブデプロイメント設定を構成する」をご参照ください。

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

    デフォルトのキャッシュサイズは 10,000 です。200,000 に増やすと、理論上のヒット率は 200000 * 50 / 100 / 100000 = 100% に上がります。

  • PartitionBy への時間ベースフィールドの追加

    日次ランキングの場合は Day フィールドを含めてください。含めない場合、State TTL により最終的な TopN 結果が破損する可能性があります。

効率的な重複排除

組み込み関数の使用における注意点

• ユーザー定義関数の組み込み関数への置換

  Realtime Compute for Apache Flink は組み込み関数を継続的に最適化しています。ユーザー定義関数よりも組み込み関数を優先してください。主な最適化は次のとおりです：

  • シリアライズ／デシリアライズのオーバーヘッドを削減します。

  • バイトレベルで直接操作します。

• KEY VALUE 関数における 1 文字区切りの使用

  KEY VALUE のシグネチャ：KEYVALUE(content, keyValueSplit, keySplit, keyName)。keyValueSplit と keySplit が 1 文字 (コロン ":"、カンマ "," など) の場合、システムは最適化アルゴリズムを使用し、content 全体を分割せずにバイナリデータ内で keyName を直接特定します。これにより、パフォーマンスが約 30% 向上します。

• LIKE 演算子の注意点

  • StartWith の場合は、LIKE 'xxx%' を使用します。

  • EndWith の場合は、LIKE '%xxx' を使用します。

  • Contains の場合は、LIKE '%xxx%' を使用します。

  • Equals の場合は、LIKE 'xxx' を使用します。これは str = 'xxx' と等価です。

  • アンダースコア (_) に一致させるには、エスケープしてください：LIKE '%seller/_id%' ESCAPE '/'。アンダースコア (_) は SQL の 1 文字ワイルドカードです。エスケープしない場合、 LIKE '%seller_id%' は seller_id、seller#id、sellerxid、seller1id に一致し、誤った結果になります。

• 正規表現関数 (REGEXP) の回避

  正規表現は非常にコストが高く、基本的な算術演算より 100 倍遅くなることもあります。また、特定の条件下では無限ループに入り、ジョブをブロックする可能性があります。Regex execution is too slow を参照してください。LIKE の使用を推奨します。正規表現関数には次が含まれます：

  • REGEXP

  • REGEXP_REPLACE

全表読み取り時の低効率とバックプレッシャーの解消

バックプレッシャーは、下流の処理が遅いことに起因する場合があります。まず、下流でバックプレッシャーが発生しているかを確認します。発生している場合は、次のいずれかの方法で解消してください：

• 並列度を上げます。

• MiniBatch などの集計最適化を有効化します (下流の集計ノード向け)。

ジョブ概要の頂点サブタスクにおける Status Durations の色の意味

ジョブの [Overview] ページでオペレーターノードをクリックし、[SubTasks] タブを選択して、[Status Durations] 列に表示される色付きの期間バッジを確認します。

Status Durations は、頂点サブタスクが各フェーズで費やした時間を示します。色の意味は次のとおりです：

• ：CREATED

• ：SCHEDULED

• ：DEPLOYING

• ：INITIALIZING

• ：RUNNING

RMI TCP Connection スレッドの概要と高 CPU 使用率の理由

CPU 使用率でソートしたスレッド監視リストでは、RMI TCP Connection(62)-172.25.240.255 スレッドが RUNNABLE ステータスで CPU 使用率 82.3% を示し、kafkaRequestSource スレッド (CPU 使用率 16.9%～26.4%、大半が TIMED_WAITING) よりもはるかに高く見えることがあります。

RMI TCP Connection スレッドは、Java の組み込み RMI (Remote Method Invocation) フレームワークに属し、リモートメソッド呼び出しを処理します。CPU 使用率は動的に変動します。短期的なスパイクは、継続的な高負荷を意味しません。時間をかけて CPU 使用率を観測してください。フレームグラフ分析 (以下) では、RMI スレッドはほとんど CPU を消費していないことが分かります。

実行中トポロジーにおける各ウォーターマークと現在時刻の差異の理由

• 理由 1：ソーステーブルのウォーターマークを TIMESTAMP_LTZ (TIMESTAMP(p) WITH LOCAL TIME ZONE) で宣言しているため、時刻に差が生じます。

  次の例では、TIMESTAMP_LTZ と TIMESTAMP 型におけるウォーターマークの動作を比較します。

  • ソーステーブルのウォーターマークに TIMESTAMP_LTZ 型を使用します。

    CREATE TEMPORARY TABLE s1 (
      a INT,
      b INT,
      ts as CURRENT_TIMESTAMP,-- CURRENT_TIMESTAMP は TIMESTAMP_LTZ を生成します。
      WATERMARK FOR ts AS ts - INTERVAL '5' SECOND 
    ) WITH (
      'connector'='datagen',
      'rows-per-second'='1',
      'fields.b.kind'='random','fields.b.min'='0','fields.b.max'='10'
    );
    CREATE TEMPORARY TABLE t1 (
      k INT,
      ts_ltz timestamp_ltz(3),
      cnt BIGINT
    ) WITH ('connector' = 'print');
    -- 結果を出力します。
    INSERT INTO t1
    SELECT b, window_start, COUNT(*) FROM
    TABLE(
        TUMBLE(TABLE s1, DESCRIPTOR(ts), INTERVAL '5' SECOND))
    GROUP BY b, window_start, window_end;

    説明

    レガシーウィンドウ構文は、TVF Window (Table-Valued Function) と同一の結果を生成します。レガシー構文の例：

    SELECT b, TUMBLE_END(ts, INTERVAL '5' SECOND), COUNT(*) FROM s1 GROUP BY TUMBLE(ts, INTERVAL '5' SECOND), b;

    Realtime Compute 開発コンソールでジョブをデプロイして実行すると、ウォーターマークと現在時刻の間に 8 時間の時間差が生じます (UTC+8 を基準)。

    • ウォーターマーク & Low Watermark

      Flink ジョブ監視 UI の [Watermarks] タブで、SubTask 0 のウォーターマーク値は 1706778525521 で、Datetime of Watermark Timestamp は 02-01 09:08:45 に対応します。ジョブ開始時刻は 02-01 17:03:04 であり、約 8 時間の差があります。左側のオペレーターパネルにも、同じ Low Watermark 02-01 09:08:45 が表示されます。

    • Task InputWatermark

      

  • ソーステーブルのウォーターマークに TIMESTAMP (TIMESTAMP(p) WITHOUT TIME ZONE) 型を使用します。

    CREATE TEMPORARY TABLE s1 (
      a INT,
      b INT,
      -- タイムゾーンなしの TIMESTAMP をシミュレートし、2024-01-31 01:00:00 から開始して秒ごとに増加させます。
      ts as TIMESTAMPADD(SECOND, a, TIMESTAMP '2024-01-31 01:00:00'),
      WATERMARK FOR ts AS ts - INTERVAL '5' SECOND 
    ) WITH (
      'connector'='datagen',
      'rows-per-second'='1',
      'fields.a.kind'='sequence','fields.a.start'='0','fields.a.end'='100000',
      'fields.b.kind'='random','fields.b.min'='0','fields.b.max'='10'
    );
    CREATE TEMPORARY TABLE t1 (
      k INT,
      ts_ltz timestamp_ltz(3),
      cnt BIGINT
    ) WITH ('connector' = 'print');
    -- 結果を出力します。
    INSERT INTO t1
    SELECT b, window_start, COUNT(*) FROM
    TABLE(
        TUMBLE(TABLE s1, DESCRIPTOR(ts), INTERVAL '5' SECOND))
    GROUP BY b, window_start, window_end;

    リアルタイムコンピューティング開発コンソールでデプロイして実行すると、Watermark は現在時刻 (具体的には、シミュレーションデータ時間) と一致し、時間差はありません。

    • ウォーターマーク & Low Watermark

      Flink Web UI のタスク詳細でオペレーター (例：GlobalWindowAggregate) を選択します。左側のオペレーター情報パネルに Low Watermark (例：01-31 01:03:49) が表示されます。右側の [Watermarks] タブに切り替えると、SubTask の Watermark 値とタイムスタンプを確認できます。いずれも時刻は一致します。

    • Task InputWatermark

      

• 理由 2: Realtime Compute 開発コンソール と Apache Flink UI のタイムゾーンの差異。

  Realtime Compute 開発コンソールでは時刻が UTC+0 で表示されるのに対し、Apache Flink UI ではブラウザのローカルタイムゾーンが使用されます。UTC+8 (北京時間) を基準にすると、Realtime Compute 開発コンソールの時刻は Apache Flink UI よりも 8 時間遅れて表示されます。

  • Realtime Compute 開発コンソール

    [ジョブ O&M] ページのトポロジーでは、ウォーターマーク時刻が UTC+0 で表示されます。たとえば、イベント時刻が北京時間の 2024/1/31 09:01:34 AM の場合、コンソールには 2024/1/31 01:01:34 AM と表示されます。

    製品コンソールのウォーターマーク関連の監視メトリクスも UTC+0 を使用するため、北京時間より 8 時間遅れて表示されます。

  • Apache Flink UI

    Apache Flink Web UI で、ジョブトポロジー内のオペレーターノード (例：GlobalWindowAggregate) を選択し、右側の [Watermarks] タブに切り替えます。SubTask のウォーターマーク値と対応するイベント時刻を確認します。たとえば、Low Watermark 1706662894000 は Datetime of Watermark Timestamp 2024/1/31 09:01:34 AM に対応します。これは処理時間ではなくイベント時間であるため、システム時刻との差があるのは正常です。

ジョブのバックプレッシャーのトラブルシューティング

1. [ジョブ O&M] ページで、対象のジョブ名をクリックすると、[概要] タブが開きます。

2. Busy と BackPressure を確認して、バックプレッシャーの発生箇所を特定します。

   Busy のインジケーターが赤いほど、タスク負荷が高いことを示します。BackPressure のインジケーターが濃いほど、バックプレッシャーの影響が強いことを示します。

   たとえば、上流オペレーターの Backpressured (max) が 99%、中間オペレーターの Busy (max) が 100% (赤で強調)、下流オペレーターの Busy (max) が 7% のみの場合、ボトルネックは中間オペレーターにあります。このオペレーターを最適化してください。

3. バックプレッシャーが発生しているオペレーターをクリックします。

4. [BackPressure] タブで、SubTask のバックプレッシャー ステータスを確認します。

   Back Pressure Status が緑の OK で、テーブルに表示される SubTasks 0～7 の Backpressured／Idle／Busy が 0%, 0%, N/A、かつすべての状態が OK の場合、ジョブにバックプレッシャーはありません。

ジョブの高レイテンシーのトラブルシューティング

[ジョブ O&M] ページで、[モニタリングとアラート] または [データカーブ] タブで currentEmitEventTimeLag および currentFetchEventTimeLag メトリックを確認します:

• currentEmitEventTimeLag が高い場合、ジョブのデータフェッチまたは処理に遅延があります。オペレーターのパフォーマンスを確認してください。

• currentFetchEventTimeLag が高い場合、遅延の原因はデータフェッチまたは上流システムの処理にあります。ネットワーク I/O と上流システムを調査してください。

データスキューに起因する Flink SQL ジョブのバックプレッシャー最適化

バックプレッシャーがデータホットスポットに起因することを (Subtask 分析で) 確認できた場合は、次の最適化手法を適用します：

• LocalGlobal の有効化 (一般的なデータホットスポット問題の解消)

  LocalGlobal メカニズムは、LocalAgg を使用してスキューしたデータを事前にローカル集計し、GlobalAgg にかかるホットスポットの負荷を軽減して、全体のパフォーマンスを向上させます。

  LocalGlobal は、単一の集計を Local と Global の 2 段階に分割します。これは MapReduce の Combine フェーズと Reduce フェーズに似ています。第 1 段階では、上流ノードがデータをローカルにバッファリングして集計 (localAgg) し、増分アキュムレーターを出力します。第 2 段階では、これらのアキュムレーターをマージ (Merge) して最終結果 (GlobalAgg) を生成します。

  • シナリオ

    標準的な集計 (SUM、COUNT、MAX、MIN、AVG など) のパフォーマンスを向上させ、これらのシナリオにおけるデータホットスポットの問題を解消します。

  • 制限事項

    LocalGlobal はデフォルトで有効ですが、以下の制限事項が適用されます：

    • MiniBatch を有効化する必要があります。

    • AggregateFunction は Merge を実装する必要があります。

  • 有効化の確認

    生成されたトポロジーに、GlobalGroupAggregate または LocalGroupAggregate という名前のノードが含まれていることを確認します。

• PartialFinal の有効化 (COUNT DISTINCT ホットスポット問題の解消)

  COUNT DISTINCT のホットスポットに対処するには、従来はクエリを手動で 2 段階集計 (modulo ベースのシャッフル層を追加) に書き換える必要がありました。Realtime Compute for Apache Flink は、PartialFinal 最適化により COUNT DISTINCT のシャッフルを自動的に実行するため、手動での書き換えは不要です。

  LocalGlobal は標準的な集計では効果的ですが、COUNT DISTINCT では効果が限定的です。ローカル集計中の distinct キーの重複排除率が低いままであるため、グローバルノードでホットスポットが継続します。

  • シナリオ

    COUNT DISTINCT が集計ノードのパフォーマンス要件を満たさない場合に使用します。

    重要

    • UDAF を含む Flink SQL では PartialFinal 最適化を使用しないでください。

    • データ量が少ない場合は PartialFinal を避けてください。不要なネットワークシャッフルが発生し、リソースを無駄にします。

  • 有効化方法

    デフォルトでは、この機能は無効になっています。この機能を有効にするには、対象のジョブの [デプロイ詳細] タブで、[ランタイムパラメーター設定] エリアの [その他の設定] セクションに次のコードを入力します。

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

  • 有効化の確認

    生成されたトポロジーの集計ステージが、単一ステージから 2 段階に変わっていることを確認します。

不安定なデータ消費レートのトラブルシューティング

考えられる原因と解決策は次のとおりです：

• 上流のデータ生成パターンが現在の処理速度と一致していません。

  上流のデータ生成パターンを分析し、生成レートと処理レートを整合させます。

• ジョブでバックプレッシャーが発生しています。

  バックプレッシャーが上流の消費に影響していないかを確認します。ジョブが 1 つのノードしか表示されない場合は、pipeline.operator-chaining: 'false' を追加してジョブを再起動し、オペレーターチェーンを分割して、消費レートに影響しているバックプレッシャーの原因ノードを特定します。

• I/O レートが異常です。

  該当時刻の Flink のデータ入力および消費レートのカーブを確認し、I/O が原因かどうかを判断します。

• 消費レートが異常です。

  消費レートの変動がガベージコレクション (GC) イベントと一致しているかを確認します。一致している場合は、TM ノードのメモリ使用量を確認してください。