差分テーブルでの低シャッフル マージの最適化

Delta Lake MERGE コマンド を使用すると、ユーザーは高度な条件でデルタ テーブルを更新できます。 MERGE コマンドを使用して、ソース テーブル、ビュー、または DataFrame のデータをターゲット テーブルに更新できます。 ただし、現在のアルゴリズムは、 変更されていない 行を処理するために完全には最適化されていません。 低シャッフル マージの最適化では、変更されていない行は、一致した行を更新するために必要なコストの高いシャッフル操作から除外されます。

低シャッフル マージが必要な理由

現在、MERGE 操作は 2 回の結合実行によって実行されます。 最初の結合では、ターゲット テーブル全体とソース データを使用して、一致した行を含むターゲット テーブルの タッチされた ファイルの一覧を検索します。 その後、2 番目の結合が実行 され、タッチされた ファイルとソース データのみが読み取られ、実際のテーブルの更新が実行されます。 最初の結合は 2 番目の結合のデータ量を減らすことですが、タッチされたファイルには変更されていない行が膨大な数存在する可能性があります。 最初の結合クエリは、指定された一致条件の列のみを読み取るので、軽くなります。 テーブル更新の 2 つ目は、すべての列を読み込む必要があり、コストのかかるシャッフル プロセスが発生します。

低シャッフル マージの最適化では、デルタは最初の結合から一致した行の結果を一時的に保持し、2 番目の結合に使用します。 結果に基づいて、変更 されていない 行が重いシャッフル プロセスから除外されます。 一致した行と変更されていない行に対して 2 つの個別の書き込みジョブが存在するため、前の動作と比較して出力ファイルの数が 2 倍になる可能性があります。 ただし、予想されるパフォーマンスの向上は、小さなファイルの問題の可能性を上回ります。

可用性

これは、Apache Spark バージョン 3.4 および 3.5 の Synapse プールで使用できます。

低シャッフル マージの利点

• タッチされたファイル内の変更されていない行は個別に処理され、実際の MERGE 操作は行われません。 MERGE の実行時間とコンピューティング リソース全体を節約できます。 多くの行がコピーされ、少数の行のみが更新されると、ゲインは大きくなります。
• 行の順序は、変更されていない行に対して保持されます。 そのため、変更されていない行の出力ファイルは、ファイルが並べ替えられた場合や Z-ORDERED の場合、データのスキップに対して効率的である可能性があります。
• MERGE 条件がタッチされたファイル内のすべての行と一致する場合でも、オーバーヘッドが小さくなります。

低シャッフル マージを有効または無効にする方法

プールまたはセッションの構成設定が完了すると、すべての Spark 書き込みパターンでこの機能が使用されます。

低シャッフル マージ最適化を使用するには、次の構成を使用して有効にします。

1. Scala と PySpark

spark.conf.set("spark.microsoft.delta.merge.lowShuffle.enabled", "true")

2. Spark SQL

SET `spark.microsoft.delta.merge.lowShuffle.enabled` = true

現在の構成値を確認するには、次に示すようにコマンドを使用します。

1. Scala と PySpark

spark.conf.get("spark.microsoft.delta.merge.lowShuffle.enabled")

2. Spark SQL

SET `spark.microsoft.delta.merge.lowShuffle.enabled`

この機能を無効にするには、次のように次の構成を変更します。

1. Scala と PySpark

spark.conf.set("spark.microsoft.delta.merge.lowShuffle.enabled", "false")

2. Spark SQL

SET `spark.microsoft.delta.merge.lowShuffle.enabled` = false