Wat gebeurt er precies wanneer je een Spark-job uitvoert?

Inleiding

Naarmate datasets groeien, kan het lastig worden om ze efficiënt op één machine te verwerken vanwege de beperkingen van de CPU-capaciteit, het geheugen en de opslagdoorvoer.

Apache Spark pakt deze uitdagingen aan met een gedistribueerde gegevensverwerkingsengine die workloads kan opsplitsen in partities en taken parallel kan uitvoeren op meerdere CPU-kernen en machines.

Spark draait echter niet simpelweg om het toevoegen van meer machines.

Inzicht in de manier waarop de onderdelen samenwerken — drivers, executors, RDD’s, DataFrames, partities, jobs, stages en taken — is essentieel voor het ontwerpen van efficiënte en betrouwbare datapijplijnen.

In dit artikel bespreek ik de architectuur van Spark, de werking van de uitvoering en de belangrijkste aandachtspunten voor prestaties aan de hand van een praktisch voorbeeld van het aggregeren van e-commerceverkopen.

1. Waarom Apache Spark? — Parallelle gegevensverwerking

Apache Spark is een engine voor gedistribueerde computing, ontworpen voor grootschalige gegevensverwerking en data-engineering.

Spark verwerkt big data met behulp van parallelle gegevensverwerking:

Figuur A. Architectuur op hoofdlijnen van een gedistribueerd Apache Spark-cluster (cluster-, driver- en workerknooppunten), met een vergelijking met uitvoering op één knooppunt (gemaakt door Kuriko IWAI)

Vergeleken met verwerking op één knooppunt (rechts in figuur A) verdeelt Spark het werk met behulp van meerdere knooppunten in taken en voert die uit op de beschikbare resources, verspreid over meerdere machines.

Gedistribueerd betekent hier niet dat meerdere machines noodzakelijk zijn. Spark kan in de lokale modus op één machine draaien. Ook dan kunnen gegevens in partities worden verdeeld en parallel worden verwerkt.

2. De kernarchitectuur van Apache Spark

Spark bestaat uit verschillende belangrijke onderdelen:

• Spark-driver (driver): het centrale proces dat een Spark-applicatie coördineert.
• Clusterbeheerder: een externe service die hardwarebronnen, zoals CPU-kernen en geheugen, toewijst.
• Workerknooppunten (worker): een fysieke of virtuele machine, of een computeknooppunt in het cluster.
• Executors: gedistribueerde workerprocessen die op workerknooppunten worden gestart.
• Java Virtual Machine (JVM): de runtime-omgeving die nodig is om de bytecode van Spark uit te voeren (omdat Spark is geschreven in Scala/Java).

Eerst coördineert de driver een Spark-applicatie en zet die om in een gerichte acyclische graaf (DAG) van fysieke uitvoeringstaken. Vervolgens plant de driver deze in op de executors.

Daarna wijst de clusterbeheerder resources toe binnen het resourcebeheersysteem van de implementatie.

Workerknooppunten leveren computecapaciteit aan executors, die taken uitvoeren en tijdens de uitvoering ervan gegevens vasthouden.

In sommige beheerde omgevingen, zoals Databricks, wordt één executor per worker gebruikt, maar dit is geen algemene regel voor Spark. Afhankelijk van de clusterconfiguratie kan een worker meerdere executors hosten.

2-1. E-commercecasus — structuur van de invoergegevens

Stel je een kleine e-commerceonderneming voor.

Deze heeft 10.000 verkooprecords, elk met een voorraadcode (SKU) en een verkochte hoeveelheid. Ga uit van drie mogelijke SKU’s: A, B, en C.

Tabel 1. Voorbeeld van een uitsplitsing van e-commercesalesrecords en de verwachte aggregaties.

Ons doel is het totale aantal geldig verkochte eenheden per SKU te berekenen:

1. Records eruit filteren waarvan qty <= 0, vervolgens
2. De records per SKU aggregeren.

Stel dat in de illustratieve volledige dataset na het filteren 9.800 van de 10.000 records overblijven. Dit zijn de eindtotalen die Spark naar verwachting oplevert:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Spark-cluster

Ik definieer een vereenvoudigd cluster voor onze casus.

Workerknooppunten (workers)

Ik stel 3 workerknooppunten in, elk met een instantietype van 8 vCPU en 32 GiB RAM:

Figuur B. Toewijzing van resources over workerknooppunten (gemaakt door Kuriko IWAI)

In totaal zijn er 24 vCPU’s beschikbaar in het systeem.

Spark-executors en toewijzing van CPU en geheugen

Stel dat ik het volgende reserveer:

• 7 executors.
• 2 vCPU-kernen per executor.
• 3 GiB JVM-heapgeheugen per executor.

De Spark-applicatie vraagt de clusterbeheerder om resources toe te wijzen aan deze 7 executors.

Vervolgens bepaalt de clusterbeheerder de toewijzing van CPU’s en geheugen op basis van het resourcegebruik van elke worker.

Zo kan deze bijvoorbeeld besluiten 14 taakslots aan te maken: 2 vCPU-kernen per executor × 7 executors:

Figuur C-1. Toewijzing van vCPU-kernen, geheugen en executors over workers, met een scheve verdeling van resources bij Worker A

Houd er rekening mee dat de 3 GiB geheugen die aan de executor is toegewezen niet het totale geheugen ervan is, omdat er ook geheugenoverhead moet worden gereserveerd voor workers en andere processen.

Als we ervan uitgaan dat één taak 1 vCPU vereist (spark.task.cpus = 1), wijst elke worker zijn vCPU’s toe:

Figuur C-2. Parallelle toewijzing van vCPU’s over workers

Dit laat twee belangrijke punten zien:

1. Elke worker heeft een eigen CPU-gebruikspercentage (in Figuur C-2 is het gebruik van Worker A 75%, terwijl dat van Workers B en C op 50% blijft).
2. Hoewel het systeem 24 vCPU-kernen heeft (8 vCPU’s per worker × 3 workers), kan het er maar 14 gebruiken; de overige 10 blijven ongebruikt voor deze taken.

2-3. De Java Virtual Machine (JVM) achter Spark

Apache Spark is gebouwd op de Java Virtual Machine (JVM).

De JVM is een runtimeomgeving die nodig is om de bytecode van Spark uit te voeren.

Omdat Spark is geschreven in Scala/Java:

• Bij gebruik van Scala of Java in de applicatie: De code wordt rechtstreeks in de JVM uitgevoerd.
• Bij gebruik van PySpark (Python): Een Python-proces voert de code uit, maar communiceert via een brug zoals Py4J met een op de JVM gebaseerde driver. De Spark-executors op de workerknooppunten draaien nog steeds in een JVM, die vervolgens Python-workerprocessen beheert om Python-code en door de gebruiker gedefinieerde Python-functies (Python UDF’s) uit te voeren.

In beide gevallen beheert de JVM de uitvoeringsthreads, de geheugentoewijzing en het garbage collection-proces voor zowel de driver als de executors.

Driver- en executor-JVM’s in een PySpark-applicatie

In een PySpark-applicatie communiceert de Python-driver met de Spark-driver die in de JVM draait;

• Python-driver: Hier wordt de PySpark-code op applicatieniveau geschreven en uitgevoerd.
• Driver-JVM: Coördineert de uitvoering en planning van Spark.
• Executor-JVM: Voert Spark-taken uit en beheert de uitvoeringsresources. Wanneer taken Python-uitvoering vereisen, worden Python-workerprocessen ingezet.
• Client-JVM (clientproces of indieningsmachine): JVM die op de lokale machine of een edge-/gatewayknooppunt draait.

Zoals te zien is in de figuur, maakt de client-JVM eerst verbinding met de driver-JVM, die vervolgens de executor-JVM’s coördineert:

Figuur D. Overzicht van de onderlinge relatie tussen client-, driver- en executor-JVM’s

Van hieruit beheert de driver-JVM de applicatie door de Spark-driver te hosten, terwijl de executor-JVM’s de afzonderlijke executors hosten die verantwoordelijk zijn voor de uitvoering van die taken.

Elke driver-JVM beschikt over JVM-heapgeheugen om gegevensverwerking, caching en gebruikersobjecten efficiënt af te handelen.

2-4. Een diepgaande blik op Resilient Distributed Datasets (RDD’s)

Een veerkrachtige gedistribueerde dataset (RDD) is een onveranderlijke, gepartitioneerde verzameling elementen die Spark parallel kan verwerken.

RDD’s houden afhankelijkheden tussen transformaties bij, waardoor Spark in veel foutscenario’s verloren gegane partities opnieuw kan berekenen.

De naam verwijst naar drie eigenschappen:

• Veerkrachtig: Spark legt de afstamming vast — welke RDD afhankelijk is van welke eerdere RDD — en kan ontbrekende partities vaak reconstrueren in plaats van replica’s van elk tussenresultaat te bewaren.
• Gedistribueerd: De dataset wordt weergegeven als partities die Spark op verschillende executors kan verwerken. Meerdere fysieke machines zijn niet vereist; ook in de lokale modus worden RDD’s ondersteund.
• Dataset: Een RDD kan sleutel-waardeparen, gehele getallen, tekenreeksen, tuples en andere ondersteunde elementen bevatten — niet alleen relationele rijen. De RDD-API’s omvatten bewerkingen zoals map, filter, reduceByKey, join, en aggregateByKey.

Een RDD is een logische verzameling en abstractie voor berekeningen, geen verklaring dat bepaalde partities permanent aan bepaalde workers zijn toegewezen.

Partities

Een partitie is een logisch gegevensblok en komt doorgaans overeen met één taak binnen een stage.

Laten we 100 partities maken (deze 100 initiële partities zijn een expliciete keuze voor dit voorbeeld).

Bij 10.000 bronelementen bevat elke partitie ongeveer 100 elementen als de invoer redelijk gelijkmatig is verdeeld:

Figuur E. Parallelle toewijzing van partities aan 100 initiële gegevenssplitsingen

3. RDD API versus DataFrame API: interne optimalisatie met Catalyst

Het oorspronkelijke RDD-voorbeeld helpt om lineage, partities en shuffle te begrijpen.

Bij typische gestructureerde ETL is de DataFrame API echter het aanbevolen uitgangspunt, omdat deze kolommen, schemagegevens, SQL-achtige expressies en bewerkingen biedt die zichtbaar zijn voor de optimizer.

Het onderstaande diagram laat de samenhang zien:

Figuur F. De samenhang tussen de grafiekconcepten van Spark: RDD-lineage, DataFrame en de uitvoerings-DAG van Spark

Net zoals een RDD RDD-lineage heeft om de transformaties vast te leggen, geeft het DataFrame-queryplan de gestructureerde query weer.

Spark stelt uitvoeringswerk samen op basis van afhankelijkheden en fysieke operators.

Bij de uitvoering van DataFrames kunnen intern RDD-gebaseerde mechanismen worden gebruikt, maar dat betekent niet dat gebruikers bij het schrijven van DataFrame-code handmatig RDD-lineage moeten opbouwen.

3-1. DataFrame-queryplan

Een DataFrame-queryplan is de reeks logische en fysieke stappen die een uitvoeringsengine genereert (het paarse vak in Figuur L).

Een DataFrame-query doorloopt verschillende representaties:

1. Logisch plan: Een beschrijving van de gevraagde bewerkingen (scannen, filteren, groeperen, aggregeren).
2. Geanalyseerd en geoptimaliseerd logisch plan: Spark lost verwijzingen op en past toepasselijke logische optimalisaties toe, zoals het verwijderen van ongebruikte kolommen of het naar voren halen van predicaten.
3. Fysiek plan: Spark kiest uitvoerbare operators, bijvoorbeeld scans, gedeeltelijke en definitieve aggregaties, uitwisselingen, joins en sorteringen.
4. Uitvoering: Het fysieke plan wordt uitgevoerd met de uitvoeringsmechanismen van Spark, waarbij uiteindelijk taken op executors worden uitgevoerd.

In onze toepassing kan het fysieke plan de volgende stappen bevatten:

• Gedeeltelijke hashaggregatie,
• Uitwisseling op basis van SKU, en
• Definitieve hashaggregatie.

3-2. Acties vergelijken: RDD API versus DataFrame API

Laten we de RDD API en de DataFrame API vergelijken aan de hand van een SKU-aggregatie.

RDD API

De elementen zijn (sku, qty) paren. Onze aggregatielogica in Python is expliciet en Spark volgt de RDD-afhankelijkheden:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

DataFrame API

Stel dat sales_df kolommen heeft sku (tekenreeks) en qty (numeriek):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

In beide voorbeelden wordt hetzelfde bedrijfsprobleem opgelost:

Ongeldige verkooprecords filteren en geldige hoeveelheden per SKU aggregeren.

En in beide gevallen collect() wordt de uitvoering van Spark gestart en worden alle resulterende records teruggegeven aan de Driver.

Het belangrijkste verschil zit hier in de representatie van het resultaat:

• RDD API: Geeft Python-elementen terug, zoals (sku, total_qty) tuples.
• DataFrame API: Geeft gestructureerde Row objecten met benoemde velden terug, zoals sku en total_qty.

Hier volgt een algemene vergelijking van de RDD- en DataFrame-API’s:

Tabel 2. Vergelijking van functies en abstractieniveau: RDD-API versus DataFrame-API (gemaakt door Kuriko IWAI)

4. Afronding

Het doel van Spark-tuning is niet om het aantal partities of het CPU-gebruik afzonderlijk te maximaliseren.

Het doel is de uitvoeringskosten en latentie van begin tot eind te verlagen zonder instabiliteit te veroorzaken.

In dit artikel leren we de belangrijkste verbanden binnen het systeem kennen:

• Resources: Workers hosten executors; executors voeren taken uit; de ingestelde CPU- en geheugencapaciteit begrenst de gelijktijdigheid.
• Programmeren met RDD’s: Transformaties leveren definities van onveranderlijke RDD’s en hun afstammingsrelaties op; acties starten de berekening.
• Programmeren met DataFrames: Transformaties leveren logische querybewerkingen op; optimalisatie kiest uitvoerbare fysieke operatoren; acties starten de berekening.
• Uitvoering: Spark verdeelt het werk in jobs, fasen en taken, en taken verwerken partities op executors. Shuffles en andere afhankelijkheden creëren coördinatiegrenzen tussen fasen.

Inzicht in de manier waarop Spark gedistribueerde workloads uitvoert, is slechts de basis.

Je kunt een rekenstroom in zeven stappen met de RDD-API leren kennen.