Effiziente Herunterskalierung und Remote-Shuffle-Manager

Gilt für:✅ Fabric Data Engineering and Data Science

Effiziente Herunterskalierung ist eine Funktion in Microsoft Fabric Spark, die Spark-Shuffle-Daten von der Lebensdauer eines Executors entkoppelt. Anstatt die Shuffle-Ausgabe auf lokalen Executor-Datenträgern zu speichern, leitet Fabric Spark Shuffle-Daten an Azure Blob Storage weiter (oder migriert sie bei Bedarf dorthin) und überlässt die Gestaltung des Schreibvorgangs der Adaptive Query Execution (AQE) selbst. Das Ergebnis ist eine schnellere Cluster-Verkleinerung, geringere Rechenkosten und ausfallsicherere Jobs – ohne dass Sie Änderungen an Ihren Abfragen, Notebooks oder Pipelines vornehmen müssen.

Overview

Effiziente Skalierung basiert auf vier kooperationsbezogenen Funktionen:

Fähigkeit Was es bewirkt
Remote Shuffle Manager (RSM) Schreibt und liest Shuffle-Daten in Azure Blob Storage anstelle von lokalen Datenträgern.
Shuffle-Migration Verschiebt Shuffle-Blöcke von einem Executor, bevor dieser außer Betrieb genommen wird, anstatt sie zu verwerfen.
Entscheidungsschicht Laufzeitrouting pro Phase, das kleine Shuffles lokal hält und große Shuffles in externen Speicher auslagert.
AQE Shuffle Write Sorgt dafür, dass Adaptive Query Execution bereits an der Shuffle-Schreibphase teilnimmt, sodass die Partitionierung von Anfang an korrekt ist.

Voraussetzungen

  • Aktivieren Sie die native Execution Engine (NEE).
  • Autoscale aktivieren (empfohlen). Effizientes Skalieren funktioniert auch ohne Autoskalierung über die später in diesem Artikel beschriebenen Spark-Konfigurationen.
  • Runtime 1.3 (Apache Spark 3.5) oder höher.

So funktioniert es

Wenn Spark eine Abfrage verarbeitet, verteilt es oft Daten zwischen den Stufen – ein Shuffle. Normalerweise speichert jeder Executor Shuffle-Daten auf seiner lokalen Festplatte, was die Executors mit diesen Daten verknüpft. Die Testamentsvollstrecker können erst freigegeben werden, wenn jeder Verbraucher das Lesen abgeschlossen hat. Diese Kopplung ist der Hauptgrund, warum Cluster nicht schnell herunterskalieren können und warum der Verlust eines Executors teure Stage-Retries verursacht.

Effizientes Herunterskalieren durchbricht diese Kopplung:

  • Große Shuffles werden über den Remote Shuffle Manager direkt in Azure Blob Storage übertragen.
  • Kleine Mischvorgänge bleiben aus Geschwindigkeitsgründen auf dem lokalen Datenträger. Wenn ihr Executor später freigegeben werden muss, verschiebt die Shuffle Migration die Blöcke zu Peers oder zu Fallback-Speicher im Hintergrund.
  • Die Entscheidungsschicht wählt zur Laufzeit pro Stufe den richtigen Pfad.
  • AQE Shuffle Write stellt sicher, dass der Writer Partitionierung erzeugt, die nachgelagertes AQE verbraucht, ohne neu zusammenzukoppeln und verschwendete E/A zu vermeiden.
                ┌───────────────────────────┐
   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

Smart Routing (Entscheidungsschicht)

Die Entscheidungsschicht evaluiert jeden Shuffle-Austausch und entscheidet:

  • Umfangreiche Shuffles → Azure Blob Storage. Maximale Skalierungs- und Fehlertoleranzvorteile.
  • Kleine Shuffle-Vorgänge → auf den lokalen Datenträger. Kein Cloud-E/A-Aufwand für winzige Übertragungen. Wenn der Executor später außer Betrieb genommen wird, übernimmt die Shuffle-Migration.

Die Entscheidungsschicht verteilt Shuffle-Daten automatisch und erfordert keine Eingabe von Ihnen. Die empfohlene Granularität ist pro Stufe.

Hauptvorteile

Niedrigere Kosten: Bezahlen Sie nur für die von Ihnen verwendete Berechnung

Beim effizienten Herunterskalieren werden Executors freigegeben, sobald ihre Arbeit erledigt ist. Sie befinden sich nicht mehr im Leerlauf, in dem Shuffle-Daten enthalten sind, die nachgelagerte Aufgaben möglicherweise schließlich lesen.

  • Schnellere Skalierung. Autoscale entfernt Knoten unmittelbar nach Abschluss des Vorgangs.
  • Weniger ungenutzte Rechenleistung. Keine „Zombie“-Executoren wurden nur dafür am Leben gehalten, ihren lokalen Shuffle zu bedienen.
  • Keine Datenträgerüberbereitstellung. Große Shuffle-Vorgänge werden in Blobspeicher ausgelagert, anstatt große lokale Datenträger zu erfordern.
  • Kosten für gebundenen Speicherplatz. Der Ausweichspeicher wird automatisch bereinigt, wenn Blöcke nicht mehr benötigt werden.

Stabilere Aufträge

Wenn "Shuffle"-Daten nur auf einem lokalen Datenträger gespeichert sind, bedeutet ein Ausführenderabsturz, dass Daten nicht mehr vorhanden sind und Spark sie erneut kompensieren muss. Bei effizienter Skalierung befinden sich Daten entweder bereits im Blob-Speicher oder werden dort migriert, bevor der Executor weggeht.

Scenario Ohne effiziente Herunterskalierung Mit effizienter Herunterskalierung
Executor stürzt ab Shuffle-Daten verloren; Stufen erneut ausgeführt Die Daten sind sicher im Speicher; keine Neukompilierung
Knotenverdrängung Daten verschwunden, teure Wiederholungen Daten bleiben erhalten; der Job läuft normal weiter.
Geordnete Außerbetriebnahme Shuffle beim Herunterfahren deaktiviert Zu Peer- oder Fallbackspeicher migrierte Blöcke
Netzwerkblips während des Abrufs Kaskadierend FetchFailedException Lesevorgänge werden aus dem Speicher gelesen, bleiben unbeeinflusst

Dieses Design beseitigt die häufigste Ursache für FetchFailedException in der Produktion.

Schnellere, wirklich elastische Skalierung

Ohne effiziente Herunterskalierung kann der Autoscaler einen Knoten nicht freigeben, solange ein Executor darauf noch Shuffle-Daten oder zwischengespeicherte Daten hält. Effiziente Skalierung entkoppelt beide:

  • Shuffle-Daten befinden sich im Blobspeicher (oder werden beim Herunterfahren dorthin migriert).
  • Der Cache bindet Executoren nicht mehr. Reproduzierbare Caches wie der Delta snapshot cache sind vom Scale-down-Schutz ausgenommen.

Der Autoscaler kann inaktive Knoten beliebig entfernen und die Clustergröße als Reaktion auf Änderungen der Arbeitslast anpassen.

Bessere Leistung bei schiefen und großen Shuffles

AQE Shuffle Write ermöglicht es Adaptive Query Execution, den Shuffle-Schreibvorgang selbst zu gestalten – indem eine Partitionierung gewählt wird, die das nachgelagerte AQE ohne erneutes Zusammenführen direkt nutzen kann, und indem weniger, besser dimensionierte Blöcke für den Remote-Speicher erzeugt werden. In Kombination mit der Entscheidungsebene erzielen Sie bei großen oder unausgewogenen Abfragen eine kürzere Gesamtlaufzeit, während die Latenz bei kleineren Abfragen unverändert bleibt.

Erste Schritte

Wenden Sie diese Konfiguration an, um den vollständigen effizienten Scaledown-Stack zu aktivieren:

# 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")

Es sind keine Codeänderungen erforderlich. Sie können diese auch in Ihren Spark-Eigenschaften in Ihrer Umgebung festlegen.

Konfigurationsreferenz

Remote Shuffle Manager (RSM)

Setting Empfohlen Was sie steuert
spark.remote.shuffle.enabled true Aktiviert die effiziente Herunterskalierung. Shuffle-Daten werden in Azure Blob Storage statt auf den lokalen Datenträgern des Executors gespeichert.

Entscheidungsschicht

Setting Empfohlen Was sie steuert
spark.sql.rsm.decisionlayer.enabled.level stage Granularität, auf der die Routenumordnung in der Entscheidungsschicht erfolgt. stage wertet jede Spark-Phase unabhängig aus.

AQE-Shuffle-Schreibvorgang

Setting Empfohlen Was sie steuert
spark.sql.adaptive.shuffleWrite.enabled true Ermöglicht AQE die Teilnahme an der Shuffle-Schreibphase. Erzeugt eine Partitionierung, die von nachgeschaltetem AQE ohne erneute Zusammenführung genutzt wird.

Note

AQE selbst (spark.sql.adaptive.enabled) muss aktiviert sein. Sie ist standardmäßig in Fabric Spark aktiviert.

Shuffle-Migration bei Außerbetriebnahme

Setting Empfohlen Was sie steuert
spark.storage.decommission.shuffleBlocks.enabled true Migriert Shuffle-Blöcke von einem Executor, der außer Betrieb genommen wird, anstatt sie zu verwerfen.
spark.storage.decommission.shuffleBlocks.cleanup true Bereinigt Shuffle-Blöcke auf dem Quell-Executor nach einer erfolgreichen Migration.
spark.storage.decommission.shuffleBlocks.migrateToFallbackStorage true Wenn kein Peer-Executor die Blöcke akzeptieren kann, werden sie in den Fallbackspeicher (Azure Blob Storage) migriert.
spark.storage.decommission.fallbackStorage.cleanUp true Entfernt Shuffle-Blöcke aus dem Fallbackspeicher, sobald sie nicht mehr benötigt werden, wodurch die Speicherkosten begrenzt werden.

Cache-bewusste dynamische Speicherzuweisung

Setting Empfohlen Was sie steuert
spark.dynamicAllocation.preventShutdownExecutorWithCache false Ermöglicht die dynamische Zuordnung zum Freigeben von Executoren auch dann, wenn sie zwischengespeicherte Blöcke enthalten.
spark.dynamicAllocation.excludeDeltaSnapshotCache true Ignoriert den Delta-Snapshot-Cache bei der Entscheidung, ob ein Executor noch einen nützlichen Cache hält. Der Delta-Momentaufnahmecache ist reproduzierbar und sollte keine Skalierung blockieren.

Erweiterte Optimierung (RSM)

Die meisten Benutzer müssen diese Standardwerte nicht ändern.

Schreibleistung

Setting Vorgabe Was sie steuert
spark.remote.shuffle.partition.buffersize 16777216 (16 MB) Puffer pro Partition vor dem Schreiben in den Speicher.
spark.remote.shuffle.blocksize 8388608 (8 MB) Größe einzelner Blöcke, die in Blob Storage hochgeladen wurden.
spark.remote.shuffle.write.maxthreads cores × 16 Maximale Anzahl von Threads zum Schreiben von Shuffle-Daten.
spark.remote.shuffle.write.maxtasks 16384 Maximale Anzahl gleichzeitiger Schreibvorgänge.

Leseleistung

Setting Vorgabe Was sie steuert
spark.remote.shuffle.read.parallel.enabled true Parallele Download-Streams beim Shuffle-Lesen.
spark.remote.shuffle.read.parallelism 4 Paralleler Downloaddatenstrom pro Aufgabe.
spark.remote.shuffle.read.prefetchqueuesize 250 Prefetch-Warteschlangentiefe während der Lesevorgänge.
spark.remote.shuffle.read.maxthreads cores × 4 Maximale Anzahl von Threads, die zum Lesen verwendet werden.

Zuverlässigkeit

Setting Vorgabe Was sie steuert
spark.remote.shuffle.retries 5 Wiederholen Sie Versuche bei vorübergehenden Speicherfehlern.
spark.remote.shuffle.retrydelayms 800 Anfängliche Wartezeit zwischen Wiederholungsversuchen.
spark.remote.shuffle.retrymaxdelayms 60000 Rückanfangskappe.

Komprimierung

Setting Vorgabe Was sie steuert
spark.remote.shuffle.compression Benutzungen spark.io.compression.codec Komprimierungsformat für Remote-Shuffle-Daten (z. B. lz4, zstd).

Leistungsergebnisse

Diagramm, das die Einsparungen bei den Rechenkosten bei aktiviertem gegenüber deaktiviertem effizientem Herunterskalieren in einem TPC-DS-Benchmark zeigt und eine Kostenreduzierung von 54 Prozent veranschaulicht.

Kosteneinsparungen berechnen (TPC-DS Benchmark)

Metric Ohne effiziente Herunterskalierung Mit effizienter Herunterskalierung
Berechnung insgesamt (VM-Minutes) 14,952 6,880
Kostensenkung — 54%

Die gesamte Auftragslaufzeit kann länger sein (Autoscale verwendet weniger gleichzeitige Executors), aber die abgerechnete Rechenleistung wird um mehr als die Hälfte reduziert.

Leistung der Entscheidungsebene (TPC-DS, RSM on)

Die Weiterleitung kleiner Shuffles auf lokale Festplatten und nur großer Shuffles auf Remote-Speicher führt gegenüber der Remote-Weiterleitung aller Shuffles zu einer Laufzeitverbesserung von bis zu 57 % – bei gleichem Scale-down-Vorteil.

Einschränkungen

  • NEE erforderlich. Effizientes Herunterskalieren hängt von der Native Execution Engine ab.
  • Nur Azure Blob Storage. Standard BlockBlobStorage mit deaktiviertem HNS. Azure Data Lake Gen2 / Konten mit aktiviertem HNS werden nicht als Remote-Shuffle-Speicher unterstützt.
  • Wird mit Azure Private Link nicht unterstützt. Umgebungen, die private Linknetzwerke verwenden, sind derzeit nicht kompatibel.
  • Die Granularität der Entscheidungsebene ist derzeit pro Stufe. Routing pro Aufgabe oder pro Partition ist nicht vorgesehen.
  • Änderung des Cacheverhaltens. Mit preventShutdownExecutorWithCache=false können Executors, die cache()/persist()-Daten vorhalten, herunterskaliert werden. Workloads, die stark vom lokalen Executor-Cache für Hot Data abhängen, sollten überprüft werden.

Verwandte Inhalte

  • Last updated on