Realtime Compute for Apache Flink:ディメンションテーブルの JOIN ステートメント

効果

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) */  // src テーブルにデータの偏りがあることを指定します。
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a