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

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

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

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

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

可用性

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

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

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