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

デプロイ構成の最適化

• リソース構成の最適化

  Ververica Platform（VVP）は、JobManager で使用できる CPU コア数と、各 TaskManager で使用できる CPU コア数に制限を設けています。使用できる CPU コアの最大数は、構成した CPU コア数と同じです。リソース構成を最適化する場合は、次の操作を実行することをお勧めします。

  • 多数のデプロイメントが並行して実行されている場合は、[Configuration] タブの [Resources] セクションでパラメーターを構成して、JobManager が使用する CPU コアの数とメモリサイズを増やすことができます。例:

    • Job Manager CPU を 4 に設定します。

    • Job Manager メモリを 8 GiB に設定します。

  • デプロイメントトポロジが複雑な場合は、[リソース] セクションの [構成] タブでパラメーターを構成して、TaskManager が使用する CPU コア数とメモリサイズを増やすことができます。例：

    • Task Manager CPU を 2 に設定します。

    • Task Manager メモリを 4 GiB に設定します。

  • taskmanager.numberOfTaskSlots のデフォルト値を保持することをお勧めします。デフォルト値は 1 です。

• スループットの向上とデータホットスポットの問題の解決

  [設定] タブの [パラメーター] セクションにある [その他の設定] フィールドに次のコードを追加します。詳細については、「コンソール操作」および「グループ集計の最適化」をご参照ください。

  execution.checkpointing.interval: 180s
  table.exec.state.ttl: 129600000
  table.exec.mini-batch.enabled: true
  table.exec.mini-batch.allow-latency: 5s
  table.optimizer.distinct-agg.split.enabled: true

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

  パラメーター	説明
  execution.checkpointing.interval	チェックポイント間隔（ミリ秒）。
  state.backend	状態バックエンドの構成。
  table.exec.state.ttl	状態データのライフサイクル（ミリ秒）。
  table.exec.mini-batch.enabled	miniBatch 機能を有効にするかどうかを指定します。
  table.exec.mini-batch.allow-latency	データを一度にエクスポートする間隔。
  table.exec.mini-batch.size	マイクロバッチ操作のためにキャッシュできるデータレコードの最大数。 説明 miniBatch 機能は、Ververica Runtime（VVR）を使用する Realtime Compute for Apache Flink で最適化されています。このパラメーターは構成しないことをお勧めします。詳細については、「主要なパラメーター」をご参照ください。
  table.optimizer.distinct-agg.split.enabled	COUNT DISTINCT 関数を使用する場合に、PartialFinal 最適化を有効にしてデータホットスポットの問題を解決するかどうかを指定します。

• 2 つのデータストリームの JOIN 操作が実行されるデプロイのパフォーマンスの向上

  SQL ストリーミングデプロイで 2 つのデータストリームを結合するために使用される JOIN 演算子を使用すると、Realtime Compute for Apache Flink エンジンは、キーと値の分離機能を有効にするかどうかを自動的に推測できます。 VVR 6.0.1 以降を使用する Realtime Compute for Apache Flink では、Realtime Compute for Apache Flink エンジンは、SQL ストリーミングデプロイの特性に基づいて、キーと値の分離機能を有効にするかどうかを自動的に推測できます。追加の構成は必要ありません。キーと値の分離機能を有効にすると、2 つのデータストリームの JOIN 操作が実行されるデプロイのパフォーマンスが大幅に向上します。一般的なシナリオでのパフォーマンステストの結果は、パフォーマンスが 40% 以上向上することを示しています。

  table.exec.join.kv-separate パラメーターを構成して、キーと値の分離機能を有効にするかどうかを指定できます。有効な値：

  • AUTO：Realtime Compute for Apache Flink エンジンは、2 つのデータストリームを結合するために使用される JOIN 演算子の状態に基づいて、キーと値の分離機能を自動的に有効にします。これがデフォルト値です。

  • FORCE：Realtime Compute for Apache Flink エンジンは、キーと値の分離機能を強制的に有効にします。

  • NONE：Realtime Compute for Apache Flink エンジンは、キーと値の分離機能を強制的に無効にします。

  説明

  キーと値の分離機能は、GeminiStateBackend でのみ有効です。

推奨される Flink SQL のプラクティス

• TopN のプラクティス

• 効率的な重複除去

• 効率的な組み込み関数

• SQL ヒント

グループ集計の最適化

• スループットを向上させるために miniBatch を有効にする

  miniBatch が有効になっている場合、Realtime Compute for Apache Flink はデータキャッシュがトリガー条件を満たしたときにデータを処理します。これにより、Realtime Compute for Apache Flink が状態データにアクセスする回数が減ります。これにより、データのスループットが向上し、データ出力が減少します。

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

  • シナリオ

    マイクロバッチ処理は、レイテンシを犠牲にしてスループットを向上させます。したがって、マイクロバッチ処理は、非常に低いレイテンシを必要とするシナリオには適用されません。ただし、データ集計シナリオでは、システムパフォーマンスを向上させるためにマイクロバッチ処理を有効にすることをお勧めします。

  • miniBatch を有効にする方法

    miniBatch 機能はデフォルトで無効になっています。この機能を有効にするには、[構成] タブの [パラメーター] セクションにある [その他の構成] フィールドに次のコードを追加する必要があります。詳細については、「コンソール操作」をご参照ください。

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

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

    パラメーター	説明
    table.exec.mini-batch.enabled	miniBatch 機能を有効にするかどうかを指定します。
    table.exec.mini-batch.allow-latency	一度にデータをエクスポートする間隔。
    table.exec.mini-batch.size	マイクロバッチ操作のためにキャッシュできるデータレコードの最大数。 説明 このパラメーターは、上記の 2 つのパラメーターと一緒に使用した場合にのみ有効になります。 miniBatch 機能は、VVR を使用する Realtime Compute for Apache Flink で最適化されています。このパラメーターは構成しないことをお勧めします。詳細については、「主要なパラメーター」をご参照ください。

• 一般的なデータホットスポットの問題を解決するために LocalGlobal を有効にする

  LocalGlobal ポリシーは、ローカル集約を使用して一部のスキューデータをフィルタリングし、グローバル集約でのデータホットスポットの問題を解決できます。これにより、デプロイメントのパフォーマンスが向上します。

  LocalGlobal ポリシーは、集計プロセスをローカル集計とグローバル集計の 2 つのフェーズに分割します。これらのフェーズは、MapReduce の combine フェーズと reduce フェーズに相当します。ローカル集計フェーズでは、Realtime Compute for Apache Flink は各アップストリームノードでローカルにキャッシュされたデータのマイクロバッチを集計し、各マイクロバッチ操作のアキュムレータ値を生成します。グローバル集計フェーズでは、アキュムレータが最終結果にマージされます。その後、グローバル集計結果が生成されます。

  • シナリオ

    LocalGlobal ポリシーは、SUM、COUNT、MAX、MIN、AVG などの一般的な集計関数を使用してデプロイメントのパフォーマンスを向上させ、データホットスポットの問題を解決したいシナリオに適しています。

  • 制限事項

    LocalGlobal はデフォルトで有効になっています。このポリシーには次の制限があります。

    • miniBatch が有効になっている場合にのみ有効になります。

    • データをマージするには、AggregateFunction を使用する必要があります。

  • 検証

    LocalGlobal が有効になっているかどうかを確認するには、最終的なトポロジに GlobalGroupAggregate ノードまたは LocalGroupAggregate ノードが存在するかどうかを確認します。

• COUNT DISTINCT 関数を使用する場合のデータホットスポットの問題を解決するために PartialFinal を有効にする

  通常の場合、COUNT DISTINCT 関数を使用する場合は、個別キーでデータを分散させるレイヤーを追加する必要があります。このように、集計プロセスを 2 つのフェーズに分割して、データホットスポットの問題を解決できます。 Realtime Compute for Apache Flink は、PartialFinal ポリシーを提供して、データを自動的に分散させ、集計プロセスを分割します。

  LocalGlobal ポリシーは、SUM、COUNT、MAX、MIN、AVG などの一般的な集計関数の パフォーマンスを向上させます。ただし、LocalGlobal ポリシーは、COUNT DISTINCT 関数のパフォーマンスの向上にはあまり効果的ではありません。これは、ローカル集計では重複する個別キーを削除できないためです。その結果、グローバル集計フェーズで大量のデータがスタックされます。

  • シナリオ

    PartialFinal ポリシーは、COUNT DISTINCT 関数を使用する場合に集計のパフォーマンスが要件を満たせないシナリオに適しています。

    説明

    • ユーザー定義集計関数（UDAF）を含む Flink SQL コードでは、PartialFinal を有効にできません。

    • リソースの無駄を防ぐために、データ量が多い場合にのみ PartialFinal を有効にすることをお勧めします。 PartialFinal はデータを自動的に分散させ、集計プロセスを 2 つのフェーズに分割します。これにより、追加のネットワークシャッフリングが発生します。

  • 有効化

    PartialFinal はデフォルトで無効になっています。この機能を有効にするには、[構成] タブの [パラメーター] セクションにある [その他の構成] フィールドに次のコードを追加する必要があります。詳細については、「コンソール操作」をご参照ください。

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

  • 検証

    最終的なトポロジで 1 層集計が 2 層集計に変更されるかどうかを確認します。

• 大量のデータが存在する場合に COUNT DISTINCT 関数を使用するシナリオでシステム パフォーマンスを向上させるために、AGG WITH CASE WHEN を AGG WITH FILTER に変更する

  統計デプロイメントは、すべてのチャネルの UV、モバイル端末の UV、PC の UV など、さまざまなディメンションでユニークビジター（UV）を記録します。多次元統計分析を実装するには、AGG WITH CASE WHEN 構文ではなく、標準の AGG WITH FILTER 構文を使用することをお勧めします。その理由は、Realtime Compute for Apache Flink の SQL オプティマイザーが filter パラメーターを分析できるためです。このように、Realtime Compute for Apache Flink は、状態データを共有することにより、異なるフィルター条件で同じフィールドに対して COUNT DISTINCT 関数を実行できます。これにより、状態データの読み取りおよび書き込み操作が削減されます。 パフォーマンス テストでは、AGG WITH FILTER 構文は AGG WITH CASE WHEN 構文よりも パフォーマンスが優れています。これは、AGG WITH FILTER 構文のデプロイ パフォーマンスが AGG WITH CASE WHEN 構文の 2 倍になるためです。

  • シナリオ

    同じフィールドで異なる条件下での結果を計算するために COUNT DISTINCT 関数を使用する場合に、AGG WITH CASE WHEN の代わりに AGG WITH FILTER を使用すると、デプロイのパフォーマンスが大幅に向上します。

  • 元のステートメント

    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

Join 最適化テクニック

• miniBatch を有効にしてスループットを向上させる

  miniBatch はデータをキャッシュしてから処理をトリガーします。バッチ内のメッセージをマージして、処理されるデータ量とステートへのアクセス回数を減らそうとします。これにより、スループットが向上し、中間出力が削減され、ダウンストリームオペレーターの負荷が軽減されます。miniBatch は、構成された最大許容レイテンシーに達した、キャッシュされたバッチサイズに達した、またはチェックポイントなどの別のイベントが発生した、のいずれかの条件が満たされたときにマイクロバッチ処理をトリガーします。

  • シナリオ

    マイクロバッチ処理は、より高いスループットと引き換えに一部のレイテンシーを犠牲にします。超低レイテンシー要件がある場合は、この機能を有効にしないでください。一般的な join シナリオ、特に次の SQL 例に示すようなカスケード外部結合を含むシナリオでは、マイクロバッチ処理を有効にすると、システムパフォーマンスが大幅に向上します。これは、アップストリームオペレーターがメッセージの折りたたみまたはマージメカニズムを使用して、ダウンストリームオペレーターへのメッセージ増幅効果を効果的に抑制するためです。

    • 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

  • 有効化する方法

    この最適化は VVR 8.0.4 以降でサポートされています。MiniBatch はデフォルトで無効になっています。MiniBatch を有効にするには、[Other Configurations] に次のコードを入力します。詳細については、「カスタムジョブ実行パラメーターを設定する方法」をご参照ください。

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

TopN の実践

• TopN アルゴリズム

  TopN の入力データストリームが、ログサービスデータソースからのデータストリームなどの静的データストリームの場合、TopN は AppendRank アルゴリズムのみをサポートします。TopN の入力データストリームが集計関数または結合関数によって処理されるデータストリームなどの動的データストリームの場合、TopN は UpdateFastRank アルゴリズムと RetractRank アルゴリズムをサポートします。UpdateFastRank のパフォーマンスは RetractRank よりも優れています。使用中のアルゴリズムの名前は、トポロジ内のノードの名前に含まれています。

  • AppendRank: 静的データストリームに対しては、このアルゴリズムのみがサポートされています。

  • UpdateFastRank: このアルゴリズムは、動的データストリームに最適です。

  • RetractRank: このアルゴリズムは、動的データストリームの基本的なアルゴリズムです。このアルゴリズムのパフォーマンスがビジネス要件を満たしていない場合は、特定のシナリオでこのアルゴリズムを UpdateFastRank に変更できます。

  次のセクションでは、RetractRank を UpdateFastRank に変更する方法について説明します。UpdateFastRank アルゴリズムを使用する場合は、次の条件が満たされていることを確認してください。

  • 入力ストリームは更新ストリームですが、DELETE (D) または UPDATE_BEFORE (UB) メッセージは含まれていません。そうでない場合、ソートフィールドの単調性が影響を受けます。対応するノードから出力されるメッセージタイプを取得するには、EXPLAIN CHANGELOG_MODE <query_statement_or_insert_statement_or_statement_set> コマンドを実行します。構文の詳細については、「EXPLAIN 文」をご参照ください。

  • 入力データストリームには、プライマリキー情報が含まれています。たとえば、GROUP BY 句を使用して、ソースのプライマリキーに基づいて列を集計します。

  • ORDER BY 句のフィールドまたは関数（ORDER BY COUNT、COUNT_DISTINCT、SUM（正の値）DESC など）の値は、ソートの逆順に単調に更新されます。

  ORDER BY SUM DESC を使用して UpdateFastRank の最適化プランを取得する場合は、正の SUM 値を取得するための条件を指定する必要があります。これにより、[total_fee] フィールドの値が正になります。

  説明

  次のサンプルコードでは、random_test テーブルには静的データストリームが含まれています。関連グループの集計結果には、DELETE メッセージまたは UPDATE_BEFORE メッセージは含まれていません。したがって、関連する集計結果フィールドの単調性は保持されます。

  RetractRank を UpdateFastRank に変更するサンプルコード:

  insert
    into print_test
  SELECT
    cate_id,
    seller_id,
    stat_date,
    pay_ord_amt-- コマンド出力には rownum フィールドは含まれていません。これにより、結果テーブルへのデータの出力を削減できます。
  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 関数の値が正であるため単調に増加します。したがって、TopN の UpdateFast アルゴリズムを使用して上位 100 件のデータレコードを取得できます。
            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;

• TopN の最適化手法

  • ランキングなしの最適化を実行する

    TopN の出力に rownum を含めないでください。結果は、フロントエンドに最終的に表示されるときにソートすることをお勧めします。これにより、結果テーブルに書き込まれるデータ量が大幅に削減されます。ランキングなしの最適化手法の詳細については、「Top-N」をご参照ください。

  • TopN のキャッシュサイズを増やす

    TopN は、状態データへのアクセス効率を向上させるために状態キャッシュを提供します。次の式を使用して、TopN のキャッシュヒット率を計算します。

    cache_hit = cache_size*parallelism/top_n/partition_key_num

    たとえば、Top100 が使用され、キャッシュには 10,000 レコードが含まれ、並列処理は 50 です。PARTITION BY 関数のキーの数が 100,000 の場合、キャッシュヒット率は 5% です。この比率は、10000 × 50/100/100,000 = 5% という数式を使用して計算されます。キャッシュヒット率が低いということは、多くのリクエストがディスクステートデータにアクセスし、ステートシークメトリックの値が安定しない可能性があることを示しています。この場合、パフォーマンスは大幅に低下します。

    したがって、PARTITION BY 関数のキーの数が多い場合は、TopN のキャッシュサイズとヒープメモリを増やすことができます。詳細については、「デプロイメントの設定」をご参照ください。

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

    この例では、TopN のキャッシュサイズをデフォルト値の 10,000 から 200,000 に増やすと、キャッシュヒット率は 100% に達する可能性があります。このキャッシュヒット率は、次の数式を使用して計算されます: 200,000 × 50/100/100,000 = 100%。

  • PARTITION BY 関数に時間フィールドを含める

    たとえば、毎日 Day フィールドをランキングに追加します。そうしないと、状態データの Time To Live（TTL）が原因で TopN の結果が乱れます。

効率的な重複除去

効率的な組み込み関数

組み込み関数を使用する場合は、次の点に注意してください。

• ユーザー定義関数 (UDF) を置き換える組み込み関数を使用する

  Realtime Compute for Apache Flink の組み込み関数は継続的に最適化されています。 可能であれば、UDF を組み込み関数に置き換えることをお勧めします。 Realtime Compute for Apache Flink は、組み込み関数を最適化するために次の操作を実行します。

  • シリアル化と逆シリアル化の効率を向上させます。

  • バイト単位のデータ操作を可能にします。

• KEYVALUE 関数で単一文字の区切り文字を使用する

  `KEYVALUE` のシグネチャは KEYVALUE(content, keyValueSplit, keySplit, keyName) です。`keyValueSplit` と `keySplit` がコロン (:) やカンマ (,) などの 1 文字である場合、システムは最適化されたアルゴリズムを使用して、コンテンツ全体を分割することなく、バイナリデータから必要な `keyName` 値を直接見つけます。これにより、パフォーマンスが約 30% 向上します。

• LIKE 演算子を使用する場合は、次の点に注意してください。

  • `StartsWith` 操作を実行するには、LIKE 'xxx%' を使用します。

  • `EndsWith` 操作を実行するには、LIKE '%xxx' を使用します。

  • `Contains` 操作を実行するには、LIKE '%xxx%' を使用します。

  • `Equals` 操作を実行するには、LIKE 'xxx' を使用します。これは str = 'xxx' と同等です。

  • アンダースコア (_) を照合するには、エスケープする必要があります: LIKE '%seller/_id%' ESCAPE '/'。アンダースコア (_) は、任意の 1 文字に一致する SQL のワイルドカード文字です。 LIKE '%seller_id%' を指定すると、クエリは seller_id だけでなく、seller#id、sellerxid、または seller1id にも一致し、誤った結果につながります。

• 正規表現の使用を避ける

  正規表現を使用したデータ処理は時間がかかり、加算、減算、乗算、除算の操作よりも数百倍高いパフォーマンスオーバーヘッドを引き起こす可能性があります。さらに、正規表現は極端な場合に無限ループに陥る可能性があります。その結果、デプロイメントがブロックされる可能性があります。デプロイメントのブロックの問題を防ぐために、LIKE オペレーターを使用することをお勧めします。一般的な正規表現の詳細については、次のリンクをクリックしてください:

  • REGEXP

  • REGEXP_REPLACE

SQLヒント

Flinkは、SQLヒント を使用してエンジンの最適化機能を向上させることをサポートしています。SQLヒントは、以下のシナリオに適しています。

• 実行計画の変更: SQLヒントを使用して、実行計画をより適切に管理できます。

• メタデータまたは統計データの追加: スキャンされたテーブルインデックスや特定のシャッフルキーに関するスキュー情報など、特定の統計データはクエリに対して動的です。プランナーから取得される計画メタデータは不正確な場合があります。SQLヒントを使用して、メタデータまたは統計データを構成できます。

• 動的テーブルオプション の構成: 動的テーブルオプションを使用すると、テーブルオプションを動的に指定またはオーバーライドできます。これらのオプションは、各クエリ内のテーブルごとのスコープで指定できます。

クエリヒントは、オプティマイザーに実行計画の変更を提案するために使用されるSQLヒントの一種です。変更は、現在のクエリヒントが配置されているクエリブロック でのみ有効になります。Flinkクエリヒントは、結合ヒントのみをサポートしています。

• 構文

  Flinkクエリヒントの構文は、Apache Calcite SQLの構文と同じです。

  # クエリヒント: // Query hints:
  SELECT /*+ hint [, hint ] */ ...
  
  hint:
          hintName '(' hintOption [, hintOption ]* ')'
  
  hintOption:
          simpleIdentifier
      |   numericLiteral
      |   stringLiteral

• 結合ヒント

  結合ヒントは、結合を動的に最適化できるクエリヒントの一種です。ディメンションテーブルの結合ヒント と 通常の結合のヒント がサポートされています。