Efficiënt schalen omlaag en shuffle-beheer op afstand

Van toepassing op:✅ Fabric Data Engineering and Data Science

Efficiënt terugschalen is een functie van Microsoft Fabric Spark die Spark-shufflegegevens ontkoppelt van de levensduur van executors. In plaats van shuffle-uitvoer vast te maken aan lokale uitvoerschijven, stuurt Fabric Spark willekeurige gegevens door naar Azure Blob Storage (of migreert deze daar op aanvraag) en kan Adaptive Query Execution (AQE) de schrijfbewerking zelf vormgeven. Het resultaat is dat clusters sneller worden afgeschaald, de compute-kosten lager zijn en jobs veerkrachtiger zijn - zonder dat je je queries, notebooks of pipelines hoeft aan te passen.

Overview

Efficiënt terugschalen is opgebouwd uit vier samenwerkende capaciteiten:

Prerequisites

• Schakel de native execution engine (NEE) in.
• Schakel autoscale in (aanbevolen). Efficiënte schaalvermindering werkt ook zonder autoscale via de Spark-configuraties die later in dit artikel worden beschreven.
• Runtime 1.3 (Apache Spark 3.5) of hoger.

Hoe werkt het?

Wanneer Spark een query verwerkt, herverdeelt het vaak data tussen fasen - een shuffle. Normaal slaat elke executor shuffle-gegevens op zijn lokale schijf, wat de executors aan die data koppelt. De executeurs kunnen pas worden vrijgegeven als elke consument klaar is met lezen. Deze koppeling is de grootste reden waarom clusters niet snel kunnen afschalen en waarom het verliezen van een executor dure stage-retries veroorzaakt.

Efficiënt terugschalen doorbreekt deze koppeling:

• Large shuffles ga rechtstreeks naar Azure Blob Storage via Remote Shuffle Manager.
• Kleine willekeurige volgordes blijven op de lokale schijf staan voor snelheid. Als de executor later moet worden vrijgegeven, verplaatst shufflemigratie de blokken op de achtergrond naar andere peers of naar fallbackopslag.
• De beslissingslaag kiest per fase tijdens runtime het juiste pad.
• AQE Shuffle Write zorgt ervoor dat het schrijfproces een partitionering oplevert die door downstream-AQE wordt gebruikt zonder opnieuw samen te voegen, zodat onnodige I/O wordt voorkomen.

                ┌───────────────────────────┐
   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

Intelligente routering (beslissingslaag)

De beslissingslaag beoordeelt elke shuffle-uitwisseling en beslist:

• Grote herschikkingen → Azure Blob Storage. Maximale schaalaanpassing en fouttolerantievoordeel.
• Kleine herschikkingen → lokale schijf. Geen cloud-I/O-overhead voor kleine overdrachten. Als de executor later buiten gebruik wordt gesteld, neemt shuffle-migratie het over.

De beslislaag routeert shufflegegevens automatisch en vereist geen invoer van je. De aanbevolen granulariteit is per fase.

Belangrijkste voordelen

Lagere kosten: betaal alleen voor de rekenkracht die u gebruikt

Met efficiënt terugschalen worden executors vrijgegeven zodra ze hun werk hebben afgerond. Ze zitten niet langer inactief met shuffle-gegevens die downstream-taken mogelijk later nog lezen.

• Sneller omlaag schalen. Met automatisch schalen worden knooppunten onmiddellijk na voltooiing van de taak verwijderd.
• Minder niet-actieve rekenkracht. Geen 'zombie'-executors die uitsluitend in leven worden gehouden om alleen hun lokale shuffle te verwerken.
• Geen schijf-overprovisioning. Grote shufflebewerkingen gaan naar blobopslag in plaats van grote lokale schijven te vereisen.
• Kosten voor gebonden opslag. Reserveopslag wordt automatisch opgeruimd wanneer de blokken niet meer nodig zijn.

Veerkrachtigere banen

Wanneer shufflegegevens alleen op de lokale schijf staan, betekent een executorcrash dat die gegevens verloren gaan en Spark ze opnieuw moet berekenen. Met efficiënte scaledown bevinden gegevens zich al in blob-opslag of worden ze daar gemigreerd voordat de uitvoerder verdwijnt.

Dit ontwerp elimineert de meest voorkomende oorzaak van FetchFailedException in productie.

Sneller, werkelijk elastische schaalbaarheid

Zonder efficiënte scaledown kan de autoscaler geen knooppunt terugwinnen zolang een executor daarop nog shufflegegevens of cachegegevens bevat. Efficiënt terugschalen ontkoppelt beide:

• Shuffle-gegevens staan in Blob Storage (of worden daarheen verplaatst bij het afsluiten).
• Cache maakt geen uitvoerders meer vast. Reproduceerbare caches, zoals delta-momentopnamecache, worden uitgesloten van scaledown-beveiliging.

De automatische schaalaanpassing kan vrijelijk niet-actieve knooppunten verwijderen en het formaat van het cluster wijzigen als reactie op wijzigingen in de werkbelasting.

Betere prestaties bij scheve en grote shuffles

AQE Shuffle Write laat Adaptive Query Execution de shuffle-write zelf vormgeven: het kiest een partitionering waar downstream-AQE zonder opnieuw samen te voegen mee verder werkt, en produceert minder blokken van een beter formaat voor externe opslag. In combinatie met de beslissingslaag krijg je snellere muurkloktijd op grote/scheve queries en ongewijzigde latentie voor kleine.

Aan de slag

Aanbevolen configuratie

Pas deze configuratie toe om de volledige efficiënte stack voor terugschalen in te schakelen:

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

Er zijn geen codewijzigingen vereist. U kunt deze ook instellen in de Spark-eigenschappen van uw omgeving.

Configuratiegids

Remote Shuffle Manager (RSM)

Beslissingslaag

AQE-shuffle-schrijfbewerking

Shufflemigratie bij buitengebruikstelling

Dynamische toewijzing op basis van cache

Geavanceerde afstelling (RSM)

De meeste gebruikers hoeven deze standaardinstellingen niet te wijzigen.

Schrijfprestaties

Leesprestaties

Reliability

Compressie

Prestatieresultaten

Kostenbesparingen berekenen (TPC-DS benchmark)

De totale taakruntime kan langer zijn (automatisch schalen maakt gebruik van minder gelijktijdige uitvoerders), maar gefactureerde rekenkracht wordt met meer dan de helft verminderd.

Prestaties van de beslissingslaag (TPC-DS, RSM on)

Het routeren van kleine shuffles naar de lokale schijf en alleen grote shuffles naar externe opslag levert een tot 57% kortere uitvoeringstijd op vergeleken met een aanpak waarbij elke shuffle naar externe opslag wordt gerouteerd, met hetzelfde voordeel bij het afschalen.

Limitations

• NEE vereist. Efficiënt omlaag schalen hangt af van de Native Execution Engine.
• Azure Blob Storage alleen. Standard BlockBlobStorage met HNS uitgeschakeld. Azure Data Lake Gen2-/HNS-accounts worden niet ondersteund als het externe shuffle-archief.
• Niet ondersteund met Azure Private Link. Omgevingen die gebruikmaken van private link-netwerken zijn momenteel niet compatibel.
• De granulariteit van de beslissingslagen is momenteel per fase. Routering per taak of per partitie valt niet binnen het bereik.
• Wijziging van cachegedrag. Met preventShutdownExecutorWithCache=false kunnen executors met cache()/persist()-gegevens worden afgeschaald. Workloads die sterk afhankelijk zijn van executor-lokale cache voor hot data moeten worden gevalideerd.

Verwante inhoud

• Systeemeigen uitvoeringsengine voor Fabric Data Engineering
• Configureer resourceprofielen in Fabric
• Overzicht van facturering voor automatisch schalen van Spark
• Apache Spark runtimes in Fabric