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?

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

1. 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).

   

2. 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()}")
   

3. 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.

Verwante inhoud

• Overzicht van best practices voor Fabric Spark
• Fabric Spark-capaciteit en planning van clusters
• Fabric Spark-beveiliging
• Ontwikkeling en bewaking