:Flink SQL コードを最適化するためのスキル

• 構文
  Flink SQL は重複排除の構文を提供していません。 指定された主キーの下で最初または最後の重複レコードを予約し、必要に応じて残りの重複レコードを破棄するには、SQL の ROW_NUMBER() ウィンドウ 関数を使用する必要があります。 この意味で、重複排除は特別な TopN 機能です。

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

  要素名	説明
  ROW_NUMBER()	行番号を計算します。 OVER 句を含むウィンドウ関数です。 値は 1 から始まります。
  PARTITION BY col1[, col2..]	オプションです。 重複レコードの主キーを格納するためのパーティション列を指定します。
  ORDER BY timeAttributeCol [asc|desc])	時間属性(proctime または rowtime) に基づいてレコードをソートするための列を指定します。 レコードは、時間属性の昇順 (Deduplicate Keep FirstRow) または降順 (Deduplicate Keep LastRow) でソートすることができます。
  rownum	行数を計算します。 この要素は次のように設定します。rownum = 1 または rownum <= 1。

  前述の構文によると、重複排除には 2 つのステップが含まれます。

  1. ROW_NUMBER() ウィンドウ関数を使用して、指定した時間属性でデータをソートし、データにランクを付けます。
     注

     • 時間属性が proctime の場合、Realtime Compute は、レコードが Realtime Compute によって処理された時間に基づいて、重複するレコードを削除します。 この場合、ランクは毎回異なる場合があります。
     • 時間属性が rowtime の場合、Realtime Compute は、レコードが Realtime Compute に書き込まれた時間に基づいて重複レコードを削除します。 この場合、ランクは毎回同じままです。
  2. 指定された主キーの下の最初のレコードを予約し、残りの重複レコードを削除します。
     注 時間属性の昇順または降順でレコードをソートすることができます。

     • Deduplicate Keep FirstRow：Realtime Compute は、時間属性の昇順でレコードをソートし、指定された主キーの下に最初のレコードを予約します。
     • Deduplicate Keep LastRow：Realtime Compute は、時間属性の降順でレコードをソートし、指定された主キーの下の最初のレコードを予約します。
• Deduplicate Keep FirstRow
  Deduplicate Keep FirstRow ポリシーを選択した場合、Realtime Compute は指定された主キーの下の最初のレコードを予約し、残りの重複レコードを破棄します。 この場合、状態データには主キー情報のみが格納され、状態データへのアクセス効率が大幅に向上します。 サンプルコードは次のとおりです。

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

  注 上記のコードは、b フィールドに基づいてテーブル T の重複レコードを削除し、システム時刻に基づいて、指定された主キーの下の最初のレコードを予約します。 この例では、proctime 属性は、Realtime Computeでレコードが処理されるシステム時間を示しています。 Realtime Compute は、この属性に基づいてテーブル T のレコードをソートします。 システム時刻に基づいて重複レコードを削除するには、proctime 属性を宣言する代わりに、PROCTIME() 関数を呼び出すこともできます。
• Deduplicate Keep LastRow
  Deduplicate Keep LastRow ポリシーを選択すると、Realtime Compute は指定された主キーの下の最後のレコードを予約し、残りの重複レコードを破棄します。 このポリシーは、パフォーマンスの点で LAST_VALUE 関数よりわずかに優れています。 Deduplicate Keep LastRow のサンプルコードは次のとおりです。

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

  注 上記のコードは、b および d フィールドに基づいてテーブル T の重複レコードを削除し、レコードが Realtime Compute に書き込まれた時間に基づいて、指定された主キーの下の最後のレコードを予約します。 この例では、rowtime 属性は、レコードが Realtime Compute に書き込まれたときのイベント時間を示しています。 Realtime Compute は、この属性に基づいてテーブル T のレコードをソートします。

• 組み込み関数を使用したユーザー定義拡張機能 (UDX) を置き換え

  Realtime Compute の組み込み関数は継続的に最適化されています。 組み込み関数を使用して、可能な限り UDX を置き換えることを推奨します。 Realtime Compute V2.0 は、以下の側面で組み込み関数を最適化します。
  • シリアル化と逆シリアル化の効率を向上させます。
  • バイトレベルでの操作が行えます。

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

  KEYVALUE 関数のシグネチャは、KEYVALUE(content, keyValueSplit, keySplit, keyName) です。 keyValueSplit および KeySplit がコロン (:) またはコンマ (,) などの単一文字の区切り文字である場合は、Realtime Compute は最適化されたアルゴリズムを使用します。 Realtime Compute は、コンテンツ全体をセグメント化する代わりに、バイナリデータの中から必要な KeyName 値を直接検索します。 これにより、ジョブのパフォーマンスが約 30% 向上します。
• 複数のキーと値のペアが存在する場合は、MULTI_KEYVALUE 関数を使用する
  注 この方法は、Realtime Compute V2.2.2 以降でサポートされています。
  クエリには、同じコンテンツに対する複数の KEYVALUE 関数が含まれる場合があります。 コンテンツに 10 個のキーと値のペアが含まれていると想定します。 これらの 10 個の値をすべて抽出してフィールドとして使用するには、10 個の KEYVALUE 関数を記述してコンテンツを 10 回解析する必要があります。 この場合、テーブル値関数 MULTI_KEYVALUE を使用することを推奨します。この関数では、コンテンツに対して 1 つの SPLIT 解析のみが必要となります。 これにより、ジョブのパフォーマンスが 50% から 100% 向上します。
• LIKE 演算子は注意して使用する
  • 指定したコンテンツで始まるレコードを照合するには、LIKE 'xxx%' を使用します。
  • 指定したコンテンツで終わるレコードを照合するには、LIKE '%xxx' を使用します。
  • 指定されたコンテンツを含むレコードを照合するには、LIKE '%xxx%' を使用します。
  • 指定されたコンテンツと同じレコードを照合するには、LIKE 'xxx' を使用します。これはstr = 'xxx' と同じです。
  • アンダースコア (_) を照合するには、LIKE '%seller/id%' ESCAPE '/ を使用します。 アンダースコア (_) は、SQL では単一文字のワイルドカードであり、任意の文字と一致できるため、エスケープされています。 LIKE '%seller_id%' を使用すると、seller_id、seller#id、sellerxid、seller1id など、多くの結果が返されます。 結果は不十分な場合があります。

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

  正規表現の実行には時間がかかる場合があり、足算、引算、乗算、除算などの他の演算と比較して、100 倍の計算リソースが必要になる場合があります。 特定の状況で正規表現を実行すると、ジョブが無限ループに陥る可能性があります。 したがって、可能な限り LIKE 演算子を使用します。 一般的な正規表現の詳細については、対応するリンクをクリックしてください。
  • REGEXP
  • REGEXP_EXTRACT
  • REGEXP_REPLACE

• キーグループ/ハッシュ：指定されたキーに基づいてデータを配布します。
• リバランス：ラウンドロビンスケジューリングを通じて各チャネルにデータを分散します。
• 動的リバランス：出力チャネルの負荷ステータスに基づいて、負荷の少ないチャネルにデータを動的に分散します。
• フォワード：プロセスがチェーンされていない場合のリバランスに似ています。 キーとチャネルがチェーンされると、Realtime Compute は指定されたキーの下のデータを対応するチャネルに配布します。
• リスケール：入力チャネルと出力チャネル間で、一対多または多対一のモードでデータを分散します。

• 動的リバランスを使用してリバランスを置き換える

  動的リバランスを使用すると、Realtime Compute は、各サブパーティションのバッファー済みデータの量に基づいて、負荷の低いサブパーティションにデータを書き込み、動的なロードバランシングを実現することができます。 静的リバランスポリシーと比較して、動的リバランスは負荷を分散し、出力計算ノードの計算容量が不均衡な場合に全体的なジョブパフォーマンスを向上させることができます。 リバランスを使用したときに出力ノードの負荷が不均衡であることがわかった場合は、動的リバランスの使用をご検討ください。 動的リバランスを使用するには、task.dynamic.rebalance.enabled パラメーターを true に設定します。 デフォルト値は false です。

• リスケールを使用してリバランスを置き換える

  注 リスケール は Realtime Compute V2.2.2 以降でサポートされています。
  5 つの並列入力ノードと 10 の並列出力ノードがあると想定します。 リバランスを使用する場合、各入力ノードはラウンドロビンスケジューリングを通じて 10 個すべての出力ノードにデータを分散します。 リスケールを使用する場合、各入力ノードは、ラウンドロビンスケジューリングを通じて 2 つの出力ノードにデータを分散するだけで済みます。 これにより、チャネル数が減少し、各サブパーティションのバッファリング速度が向上するため、ネットワーク効率が向上します。 入力データが偶数で、並列入力ノードと並列出力ノードの数が同じ場合、リスケールを使用してリバランスを置き換えることができます。 リスケールを使用するには、enable.rescale.shuffling パラメーターを true に設定します。 デフォルト値は false です。

# Exactly-once semantics.
blink.checkpoint.mode=EXACTLY_ONCE
# The checkpoint interval, in milliseconds.
blink.checkpoint.interval.ms=180000
blink.checkpoint.timeout.ms=600000
# Realtime Compute V2.X uses Niagara as the state data backend, and uses it to set the lifecycle (in milliseconds) of the state data.
state.backend.type=niagara
state.backend.niagara.ttl.ms=129600000
# Realtime Compute V2.X enables micro-batch processing with an interval of 5 seconds. (You cannot set this parameter when you use a window function.)
blink.microBatch.allowLatencyMs=5000
# The allowed latency for a job.
blink.miniBatch.allowLatencyMs=5000
# The size of a batch.
blink.miniBatch.size=20000
# Enable local aggregation. This feature is enabled by default in Realtime Compute V2.X, but you must manually enable it if you use Realtime Compute V1.6.4.
blink.localAgg.enabled=true
# Enable PartialFinal to resolve data hotspot issues when you run the CountDistinct function in Realtime Compute V2.X.
blink.partialAgg.enabled=true
# Enable UNION ALL for optimization.
blink.forbid.unionall.as.breakpoint.in.subsection.optimization=true
# Enable OBJECT REUSE for optimization.
#blink.object.reuse=true
# Configure garbage collection for optimization. (You cannot set this parameter when you use a Log Service source table.)
blink.job.option=-yD heartbeat.timeout=180000 -yD env.java.opts='-verbose:gc -XX:NewRatio=3 -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:ParallelGCThreads=4'
# Set the time zone.
blink.job.timeZone=Asia/Shanghai