効率的なスケールダウンとリモート シャッフル マネージャー

対象:✅ ファブリックデータエンジニアリングおよびデータサイエンス

効率的なスケールダウンは、Spark シャッフル データを Executor の有効期間から切り離す、Microsoft Fabric Spark の機能です。 ローカルの Executor ディスクにシャッフル出力をピン留めする代わりに、Fabric Spark はデータのシャッフルをAzure Blob Storageにルーティング (または必要に応じて移行) し、アダプティブ クエリ実行 (AQE) が書き込み自体を形成できるようにします。 その結果、クラスタのスケールダウンが速くなり、計算コストが低くなり、クエリやノートブック、パイプラインに変更が加わらず、よりレジリエントなジョブが得られます。

Overview

効率的なスケールダウンは、次の 4 つの連携機能から構築されます。

Prerequisites

• ネイティブ実行エンジン(NEE)を有効にしてください。
• 自動スケールを有効にしてください(推奨)。 効率的なスケーリングダウンは、この記事の後半で説明するSpark構成を通じてオートスケールなしでも機能します。
• ランタイム 1.3 (Apache Spark 3.5) 以降。

どのように機能するのか

Sparkがクエリを処理する際、ステージ間でデータを再分配することがよくあり、これをシャッフルと呼びます。 通常、各エグゼキューターはローカルディスクにシャッフルデータを保存し、そのデータにエグゼキュータを紐付けます。 すべての消費者が読み終えるまで、遺言執行者は解放できません。 この結合がクラスタが迅速にスケールダウンできず、エグゼキュータを失うと高コストなステージ再試行が発生する最大の理由です。

効率的なスケールダウンにより、この結び付きは解消されます。

• Large shufflesリモート シャッフル マネージャーを介して直接Azure Blob Storageに移動します。
• 小さなシャッフルは、高速化のためローカルディスク上に保持されます。 もし後で実行者を解放する必要がある場合、シャッフル移行によってブロックはバックグラウンドのピアやフォールバックストレージに移動されます。
• 決定層は実行時に各段階で正しい経路を選択します。
• AQE シャッフル書き込み により、ダウンストリーム AQE が再結合せずに消費するパーティション分割がライターによって生成され、無駄な I/O が回避されます。

                ┌───────────────────────────┐
   Query  ───►  │   AQE + decision layer    │   per-stage choice
                └─────────────┬─────────────┘
                              │
                ┌─────────────▼─────────────┐
                │   AQE Shuffle Write       │   partition-aware writer
                └─────┬─────────────────┬───┘
                      │                 │
              local   ▼                 ▼   remote
        ┌────────────────────┐   ┌──────────────────┐
        │  Local disk +      │   │  RSM → Azure     │
        │  shuffle migration │   │  Blob Storage    │
        └─────────┬──────────┘   └─────────┬────────┘
                  │ on decommission        │
                  ▼                        ▼
        fallback storage   Remote shuffle store

スマートルーティング(意思決定層)

意思決定層は各シャッフル交換を評価し、以下を決定します:

• 大規模なシャッフル → Azure Blob Storage 最大スケールダウンとフォールト トレランスの利点。
• 小規模なシャッフル → ローカル ディスク 小さな転送に対するクラウド I/O オーバーヘッドはありません。 executor が後で停止された場合は、シャッフルマイグレーションが引き継ぎます。

意思決定レイヤーはシャッフルデータを自動でルーティングし、あなたの入力は一切必要ありません。 推奨される粒度はステージ単位です。

主な利点

コストの削減: 使用するコンピューティングに対してのみ支払う

効率的なスケールダウンにより、Executor は作業が完了するとすぐにリリースされます。 下流タスクが後で読み取る可能性のあるシャッフルデータを保持したまま、何もせず遊休状態になることはなくなります。

• より高速なスケールダウン。 自動スケールでは、タスクの完了後すぐにノードが削除されます。
• アイドル状態のコンピューティングを削減。 ローカルシャッフルを提供するためだけに生かされ続ける「ゾンビ」エグゼキューターはありません。
• ディスクオーバープロビジョニングなし。 大規模なシャッフル処理は、大容量のローカル ディスクを必要とせず、Blob Storage に格納されます。
• 有界ストレージ コスト。 フォールバック ストレージは、ブロックが不要になったときに自動的にクリーンアップされます。

回復性の高いジョブ

シャッフル データがローカル ディスクにのみ存在する場合、Executor のクラッシュはデータがなくなったことを意味し、Spark はそれを再計算する必要があります。 効率的なスケールダウンにより、データは既に BLOB ストレージに格納されているか、Executor が終了する前にそこに移行されます。

この設計により、生産中の最も一般的な FetchFailedException 原因が排除されます。

迅速で、本当に柔軟なスケーリング

効率的なスケールダウンがないと、オートスケーラーはノードを再利用できませんが、その上の実行プログラムはシャッフル データまたはキャッシュされたデータを保持します。 効率的なスケールダウンでは、その両方を切り離せます：

• シャッフル データは BLOB ストレージ内にあります (または、シャットダウン時にそこに移行されます)。
• キャッシュはエグゼキューターを固定しなくなりました。 Delta スナップショットキャッシュなどの再生成可能なキャッシュは、スケールダウン保護の対象外です。

自動スケーラーは、ワークロードの変更に応じて、アイドル状態のノードを自由に削除し、クラスターのサイズを変更できます。

偏りのあるシャッフルや大規模なシャッフルでのパフォーマンス向上

AQE Shuffle Writeは、Adaptive Query Executionがシャッフル書き込み自体を形作ることを可能にします。つまり、下流のAQEが再統合せずに消費するパーティションを選び、リモートストレージ用のブロックを少なく、より大きなブロックを生成することです。 決定レイヤーと組み合わせることで、大きくて歪んだクエリではウォールクロック時間が速くなり、小さなクエリではレイテンシが変わりません。

概要

推奨される構成

この構成を適用して、完全に効率的なスケールダウン スタックを有効にします。

# Remote Shuffle Manager
spark.conf.set("spark.remote.shuffle.enabled", "true")

# Decision layer — per-stage routing of local vs. remote shuffle
spark.conf.set("spark.sql.rsm.decisionlayer.enabled.level", "stage")

# AQE participates in shuffle write
spark.conf.set("spark.sql.adaptive.shuffleWrite.enabled", "true")

# Shuffle migration on executor decommission
spark.conf.set("spark.storage.decommission.shuffleBlocks.enabled", "true")
spark.conf.set("spark.storage.decommission.shuffleBlocks.cleanup", "true")
spark.conf.set("spark.storage.decommission.shuffleBlocks.migrateToFallbackStorage", "true")
spark.conf.set("spark.storage.decommission.fallbackStorage.cleanUp", "true")

コードに変更を加える必要はありません。 これらは、環境の Spark プロパティでも設定できます。

構成参照

リモート シャッフル マネージャー (RSM)

意思決定層

AQE シャッフル書き込み

廃止時のシャッフル移行

キャッシュ対応の動的割り当て

高度なチューニング (RSM)

ほとんどのユーザーは、これらの既定値を変更する必要はありません。

書き込み性能

読み取り性能

Reliability

Compression

パフォーマンスの結果

コンピューティング コストの削減 (TPC-DS ベンチマーク)

ジョブの総実行時間は長くなる可能性があります（自動スケールでは同時実行エグゼキューター数が少なくなるため）が、課金対象のコンピュートは半分以上削減されます。

意思決定層のパフォーマンス(TPC-DS、RSM オン)

小さなシャッフルはローカルディスクにルーティングし、大きなシャッフルはリモートストレージにルーティングすることで、すべてのシャッフルをリモートでルーティングするのと比べて最大57% の実行時間向上を実現し、同じスケールダウン効果が得られます。

制限事項

• NEE が必要です。 効率的なスケールダウンは、ネイティブ実行エンジンによって異なります。
• Azure Blob Storageのみ。 Standard BlockBlobStorage（HNS は無効）。 Azure Data Lake Gen2/HNS 対応アカウントは、リモート シャッフル ストアとしてサポートされていません。
• Azure Private Linkではサポートされていません。 プライベート リンク ネットワークを使用する環境には、現在互換性がありません。
• 意思決定層の細分 度は現在、各段階ごとに分類されています。 タスクごとまたはパーティションごとのルーティングはスコープ内にありません。
• キャッシュ動作の変更。 preventShutdownExecutorWithCache=falseでは、cache()/persist()データを保持している Executor がスケールダウンされる可能性があります。 ホット データの Executor ローカル キャッシュに大きく依存するワークロードは検証する必要があります。

関連するコンテンツ

• Fabric Data Engineering のネイティブ実行エンジン
• Fabricでリソースプロファイルを設定する
• Spark 自動スケーリングの課金概要
• Fabric の Apache Spark Runtime