Basisbeginselen van Spark

Basisconcepten die ten grondslag liggen aan grootte, optimalisatie en probleemoplossing. Lees dit eerst als u nog niet eerder met Spark in Fabric werkt.

Algemene dos en don'ts

Scenario: U bent nieuw voor Spark. Wat zijn de do's en don'ts?

Gebruiksituatie	Beste praktijken
Geoptimaliseerde geserialiseerde indelingen gebruiken	Do: Geef de voorkeur aan indelingen zoals Avro, Parquet of Optimized Row Columnar (ORC) omdat ze een schema insluiten, compact zijn en opslag en verwerking optimaliseren. Gebruik in Fabric de Delta-indeling voor atomiciteit, consistentie, isolatie, duurzaamheid (ACID) garanties en prestatievoordelen
Wees voorzichtig met XML/JSON	Vertrouw niet op schemadeductie voor grote JSON-bestanden (JavaScript Object Notation) of Extensible Markup Language (XML), omdat Spark de volledige gegevensset leest om het schema af te leiden, waardoor de verwerking wordt vertraagd en geheugen intensief verbruikt. Geef een statisch primair schema op bij het lezen van JSON/XML of gebruik `.option("samplingRatio", 0.1)` om leesbewerkingen te versnellen, maar houd er rekening mee dat als het voorbeeld niet de volledige gegevensset vertegenwoordigt, leesbewerkingen mogelijk mislukken. Bij een veiligere benadering wordt het schema afgeleid van een representatieve steekproef en blijft het schema behouden voor alle leesbewerkingen. Vermijd het parseren van grote XML-bestanden. XML-parsering wordt inherent langzamer uitgevoerd vanwege tagverwerking en typecasting.
Join-verbindingen optimaliseren en filtercriteria toepassen	Doe: Pas kolomuitsnoeiing en rijniveau-filtering toe voordat je joins uitvoert om shuffles en geheugengebruik te verminderen. De Catalyst Optimizer verwerkt predicaat pushdown automatisch wanneer u DataFrame-API's gebruikt. Vermijd RDD-API's (Resilient Distributed Dataset) omdat ze Katalysator-optimalisaties omzeilen.
Geef de voorkeur aan DataFrames boven RDD's	Do: DataFrames gebruiken in plaats van RDD's voor de meeste bewerkingen. DataFrames maken gebruik van de Catalyst Optimizer- en Tungsten-uitvoeringsengine voor efficiënte uitvoering.
Adaptieve queryuitvoering inschakelen (AQE)	Do: Schakel AQE in om partities in willekeurige volgorde dynamisch te optimaliseren en scheefgetrokken gegevens automatisch te verwerken.

Geheugenbeheer van executor

Scenario: U wilt inzicht hebben in het geheugenbeheer van de uitvoerders voor het afstemmen van de prestaties.

Zelfs als een uitvoerder is geconfigureerd met 56 GB geheugen, staat Spark niet toe dat al deze rechtstreeks voor gebruikersgegevens worden gebruikt. Spark Core verdeelt en beheert uitvoerdersgeheugen:

Gereserveerd geheugen: Een vast gedeelte dat is gereserveerd voor interne overhead van het systeem en Spark (bijvoorbeeld Java Virtual Machine (JVM), interne elementen.
Gebruikersgeheugen: Slaat door de gebruiker gedefinieerde functies (UDF's), lokale variabelen, gegevensstructuren (lijsten, kaarten, woordenlijsten) en objecten op die tijdens de berekening zijn gemaakt.
Opslaggeheugen: Bevat gegevens in de cache, persistente gegevens, broadcastvariabelen en shuffle-gegevens die in de cache kunnen worden opgeslagen.
Uitvoeringsgeheugen: Gebruikt voor tussentijdse berekeningen zoals herschikkingen, koppelingen, sorteringen en aggregaties.
Dynamisch geheugen delen: De grens tussen opslag- en uitvoeringsgeheugen kan worden verplaatst. Spark kan geheugen lenen van de ene regio naar de andere, waardoor flexibel geheugengebruik mogelijk is.
Spill: Treedt op wanneer de vraag naar opslag- of uitvoeringsgeheugen groter is dan het beschikbare geheugen na het uitlenen. Dit dwingt gegevens naar schijf, wat van invloed kan zijn op de prestaties.

Fouten bij Out of Memory (OOM)

Scenario: Spark-taken mislukken met OOM-fouten (Onvoldoende geheugen).

Driver OOM:

OOM-fouten van stuurprogramma's treden op wanneer het Toegewezen geheugen van het Spark-stuurprogramma wordt overschreden.

Veelvoorkomende oorzaak: driver-intensieve bewerkingen, zoals collect(), countByKey(), of grote toPandas() aanroepen die teveel gegevens in het geheugen van het stuurprogramma ophalen.

Risicobeperking: Vermijd waar mogelijk chauffeur-zware bewerkingen. Als dit onvermijdelijk is, vergroot de besturingsprogramma-grootte en voer een benchmarktest uit om de optimale configuratie te vinden.

Uitvoerder onvoldoende geheugen (OOM):

OOM-fouten van executor treden op wanneer een Spark-uitvoerprogramma het toegewezen geheugen overschrijdt.

Veelvoorkomende oorzaak: geheugen- en rekenintensieve transformaties voor grote gegevenssets (bijvoorbeeld brede joins, aggregaties, shuffles) of gegevenssets in de cache/persistente gegevenssets die het beschikbare geheugen van de uitvoerder overschrijden (uitvoering en opslagregio's).

Risicobeperking: Verhoog indien nodig het geheugen van de executor, stem de Spark-geheugenfracties (spark.memory.fraction, spark.memory.storageFraction) af en kies ervoor om selectief te bewaren. Zorg ervoor dat in de cache opgeslagen gegevens binnen het beschikbare geheugen passen.

Gegevensscheefheid

Symptomen van scheefheid:

Een paar taken duren langer dan andere taken in de Spark-gebruikersinterface (fasetaken tonen zware staart).
Grote tussenruimte tussen mediaan en maximale taaktijden in metrische fasegegevens.
Fasen met grote shuffle lees- of schrijfgrootten voor enkele partities.

Veelvoorkomende oorzaken:

Ongelijke gegevensdistributie voor de join- en groepssleutels (hot keys).
Onjuiste partitionering of te weinig partities voor het gegevensvolume.
Upstream-gegevensafwijkingen die grote records of veel null/lege sleutels produceren.

Mitigatie

Herpartitioneren of samenvoegen om de paralleliteit van de partitie en de balans in grootte te vergroten.
Pas sleutel-"salting" of aangepaste partitionering toe om veelgebruikte sleutels over partities te verdelen.
Gebruik AQE (Adaptive Query Execution) om partities na de shuffle samen te voegen en optimalisaties voor scheve joins in te schakelen.
Gebruik broadcast-joins voor kleine opzoektabellen om herschikkingen volledig te vermijden.
Behoud evenwichtige tussenliggende gegevenssets vóór dure fasen en voer de taak opnieuw uit.

Aanbevolen procedures voor UDF

Scenario: U moet aangepaste logica toepassen die niet kan worden uitgedrukt via ingebouwde DataFrame-functies.

Gebruik waar mogelijk Spark DataFrame-API's. De Catalyst Optimizer optimaliseert ingebouwde functies en voert ze systeemeigen uit op de JVM, zodat ze de beste prestaties leveren.

Als u een UDF (door de gebruiker gedefinieerde functie) moet gebruiken, vermijdt u reguliere PySpark Python UDF's. Overweeg in plaats daarvan de volgende alternatieven:

Pandas UDF's (ook wel vectorized UDF's genoemd): Gebruik Apache Arrow voor efficiënte gegevensoverdracht tussen JVM en Python. Pandas UDF's staan vectorgebaseerde bewerkingen toe, waardoor de prestaties aanzienlijk verbeteren vergeleken met Python UDF's die rij voor rij werken.
Scala/Java UDF's: rechtstreeks uitvoeren op de JVM, waardoor de overhead van Python-serialisatie wordt vermeden. Scala/Java UDF's presteren doorgaans beter dan Python UDF's.

Wees voorzichtig met Python UDF's. Elke uitvoerder start een afzonderlijk Python-proces, waarvoor serialisatie en deserialisatie van gegevens tussen de JVM en Python vereist is. Hierdoor ontstaat een prestatieknelpunt, met name op schaal.

Foutlogboekregistratie

Scenario: Aanbevolen procedures voor foutlogboekregistratie in Fabric Spark

Gebruik log4j in plaats van print() welke de bestuurder zwaar belast. Met log4j kunt u toegang krijgen tot de stuurprogrammalogboeken en ze doorzoeken (met behulp van de logboeknaam, bijvoorbeeld: PySparkLogger).
Lees-, schrijf- en transformatiebewerkingen omsluiten met try- en except-blokken. Gebruiken logger.error voor uitzonderingen en logger.info voor voortgangsberichten.
- Python-logboekregistratie: Ideaal voor logboekregistratiebewerkingen, statusupdates of foutopsporingsgegevens uit code die alleen op het Spark-stuurprogramma worden uitgevoerd. De logboekregistratiemodule van Python wordt niet doorgegeven aan verwerkerlogboeken. Raadpleeg de documentatie voor notebooks ontwikkelen, uitvoeren en beheren.
- Spark-logboek4j: De standaard voor robuuste toepassingslogboeken op productieniveau in Spark, omdat deze systeemeigen kan worden geïntegreerd met stuurprogramma-/uitvoerderslogboeken van Spark.
Voorbeeld van log4j-gebruik in PySpark:
```
import traceback
# Get log4j logger
log4jLogger = spark._jvm.org.apache.log4j
logger = log4jLogger.LogManager.getLogger("PySparkLogger")
logger.info("Application started.")
try:
    # Create DataFrame with 20 records
    data = [(f"Name{i}", i) for i in range(1, 21)]  # 20 records
    df = spark.createDataFrame(data, ["name", "age"])
    logger.info("DataFrame created successfully with 20 records.")
    df.show(s)  # 's' is not defined -> will throw error but the application will not fail
except Exception as e:
    logger.error(f"Error while creating or showing DataFrame: {str(e)}\n{traceback.format_exc()}")
```
Foutbewaking centraliseren:
- Gebruik de diagnostische emitterextensie (Apache Spark-toepassingen bewaken met Azure Log Analytics) in de omgeving en koppel deze aan de Notebooks waarop Spark-toepassingen worden uitgevoerd. De emitter kan gebeurtenislogboeken, aangepaste logboeken (zoals log4j) en metrische gegevens verzenden naar Azure Log Analytics/Azure Storage/Azure Event Hubs. Geef de log4j-naam door aan de eigenschap: spark.synapse.diagnostic.emitter.\<destination\>.filter.loggerName.match.
- Daarnaast kunt u voor debuggen ook mislukte rijen/records verzamelen in Lakehouse-tabellen (LH) om onjuistheden op recordniveau vast te leggen.

Feedback

Is deze pagina nuttig?

Last updated on 2025-10-23