Realtime Compute for Apache Flink:ルックアップ結合

効果

SHUFFLE_HASH 結合戦略では、JOIN 操作が実行される前に、主流データを結合キーに基づいてシャッフルできます。キャッシュポリシーが LRU の場合、キャッシュヒット率が向上し、I/O リクエストの数が減少します。キャッシュポリシーが ALL の場合、メモリ使用量が削減されます。各 SHUFFLE_HASH 結合ヒントで複数のディメンションテーブルを指定できます。

制限事項

SHUFFLE_HASH 結合戦略を使用すると、メモリのオーバーヘッドは削減されます。しかし、上流のデータを結合キーに基づいてシャッフルする必要があるため、追加のネットワークオーバーヘッドが発生します。したがって、SHUFFLE_HASH 結合戦略は次のシナリオには適していません：

• 主流データの結合キーに深刻なデータスキューがあります。SHUFFLE_HASH 結合戦略を使用してデータを結合すると、結合演算子がデータスキューのためにパフォーマンスボトルネックを引き起こす可能性があります。これにより、ストリーミングデプロイメントでは深刻なバックプレッシャーが、バッチデプロイメントでは深刻なロングテールが発生する可能性があります。このシナリオでは、REPLICATED_SHUFFLE_HASH 結合戦略を使用することを推奨します。

• ディメンションテーブルのデータ量が少なく、キャッシュポリシーが ALL の場合にテーブルのロード中にメモリボトルネックが発生しない場合、SHUFFLE_HASH 結合戦略で節約されるメモリオーバーヘッドは、SHUFFLE_HASH 結合戦略を使用することで発生する追加のネットワークオーバーヘッドと比較して、費用対効果が高くない可能性があります。

サンプルコード

-- dim1 ディメンションテーブルに対してのみ SHUFFLE_HASH 結合戦略を有効にします。
SELECT /*+ SHUFFLE_HASH(dim1) */
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b

-- dim1 と dim2 の両方のディメンションテーブルに対して SHUFFLE_HASH 結合戦略を有効にします。
SELECT /*+ SHUFFLE_HASH(dim1, dim2) */
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b

-- ヒントで dim1 ディメンションテーブルのエイリアス D1 を使用して、SHUFFLE_HASH 結合戦略を有効にします。
SELECT /*+ SHUFFLE_HASH(D1) */
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b

-- ヒントで dim1 と dim2 のディメンションテーブルのエイリアスを使用して、SHUFFLE_HASH 結合戦略を有効にします。
SELECT /*+ SHUFFLE_HASH(D1, D2) */
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b

効果

REPLICATED_SHUFFLE_HASH の効果は、基本的に SHUFFLE_HASH の効果と同じです。ただし、REPLICATED_SHUFFLE_HASH は、同じキーを持つ主流データを指定された数の並行スレッドにランダムに分散させることで、データスキューによるパフォーマンスボトルネックを解決します。各 REPLICATED_SHUFFLE_HASH 結合ヒントで複数のディメンションテーブルを指定できます。

制限事項

• table.exec.skew-join.replicate-num パラメーターを設定して、スキューデータを含むバケットの数を指定する必要があります。このパラメーターのデフォルト値は 16 です。このパラメーターの値は、ディメンションテーブルの結合演算子の並行スレッド数より大きくすることはできません。このパラメーターの設定方法の詳細については、「ジョブのカスタム実行パラメーターを設定する方法」をご参照ください。

• 更新ストリームはサポートされていません。主流が更新ストリームであり、REPLICATED_SHUFFLE_HASH 結合戦略を使用すると、エラーが返されます。

サンプルコード

-- dim1 ディメンションテーブルに対して REPLICATED_SHUFFLE_HASH 結合戦略を有効にします。
SELECT /*+ REPLICATED_SHUFFLE_HASH(dim1) */ 
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a

-- ヒントで dim1 ディメンションテーブルのエイリアスを使用して、REPLICATED_SHUFFLE_HASH 結合戦略を有効にします。
SELECT /*+ REPLICATED_SHUFFLE_HASH(D1) */ 
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a

効果

指定されたテーブルにデータスキューがある場合、オプティマイザーはディメンションテーブルの JOIN 操作に REPLICATED_SHUFFLE_HASH 結合戦略を使用します。SKEW は単なる糖衣構文であり、実際には下位レイヤーで REPLICATED_SHUFFLE_HASH 結合戦略が使用されます。

制限事項

• 各 SKEW ヒントでは、1 つのテーブルのみを指定できます。

• テーブル名は、ディメンションテーブルではなく、データスキューのあるプライマリテーブルの名前でなければなりません。

• 更新ストリームはサポートされていません。主流が更新ストリームであり、SKEW 結合戦略を使用すると、エラーが返されます。

サンプルコード

SELECT /*+ SKEW(src) */  
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a