PySpark

Introductie

Python for Data Engineers sloot af met de grens van wat één machine comfortabel aankan. PySpark is Python's interface naar Apache Spark, het gedistribueerde verwerkingsraamwerk dat werk over meerdere machines verdeelt zodra data te groot wordt voor één machine — of zelfs wanneer dat niet het geval is, simpelweg omdat parallellisatie sneller is. Eerdere hoofdstukken toonden al PySpark-syntax (Medallion Architecture, Databricks); dit hoofdstuk legt uit waarom die code zich gedraagt zoals hij doet — de mechaniek onder de syntax, zonder welke PySpark-performance-problemen onbegrijpelijk blijven.

Lazy Evaluation: Transformations versus Actions

Het eerste en meest verwarrende concept voor nieuwkomers: PySpark-code voert niets uit op het moment dat je hem schrijft. Operaties zijn onderverdeeld in twee categorieën.

Transformations (filter, select, withColumn, groupBy, join) beschrijven wat er moet gebeuren, maar voeren niets uit — ze bouwen alleen een uitvoeringsplan op, een DAG van bewerkingen, precies zoals dbt's ref()-graph uit dbt Fundamentals declareert wat moet gebeuren zonder het meteen uit te voeren.

Actions (count, collect, show, write) triggeren de daadwerkelijke uitvoering — pas op dat moment analyseert Spark het volledige opgebouwde plan en voert het uit.

df_filtered = df.filter(col("amount") > 0)      # transformation: gebeurt nog niets
df_grouped = df_filtered.groupBy("category").sum("amount")  # nog steeds niets
df_grouped.show()                                 # ACTION: nu pas voert Spark alles uit

Dit verklaart een veelvoorkomende verwarring: code die een filter bevat met een fout erin, geeft pas een foutmelding bij de show() op de laatste regel, niet bij de filter() zelf — Spark heeft tot dat moment alleen een plan gebouwd, nooit data aangeraakt. Het verklaart ook waarom df.count() twee keer aanroepen in dezelfde sessie de hele berekening twee keer opnieuw uitvoert, tenzij je expliciet cachet (verderop in dit hoofdstuk) — lazy evaluation onthoudt geen tussenresultaten uit zichzelf.

Partities, Executors en Tasks: Hoe Werk Wordt Verdeeld

Een Spark DataFrame is logisch verdeeld in partities — stukken van de data die onafhankelijk van elkaar verwerkt kunnen worden. Een cluster bestaat uit meerdere executors (worker-processen, elk met eigen CPU-cores en geheugen, verdeeld over de machines uit Databricks of een vergelijkbaar platform), en Spark's scheduler wijst elke partitie toe als een task aan een beschikbare executor-core. Meer partities dan cores betekent dat sommige tasks moeten wachten; te weinig partities (bijvoorbeeld één partitie voor een enorme dataset) betekent dat maar één core werkt terwijl de rest van het cluster stilstaat — een directe, praktische reden om partitieaantal bewust te beheren, uitgewerkt verderop.

print(df.rdd.getNumPartitions())   # hoeveel stukken is deze DataFrame nu verdeeld?

Dit is de PySpark-specifieke uitwerking van het MPP-principe (Massively Parallel Processing) uit Data Warehousing Fundamentals — dezelfde onderliggende gedachte (werk verdelen over meerdere rekenknopen), hier zichtbaar en direct beïnvloedbaar op DataFrame-niveau in plaats van verborgen achter een warehouse se automatische micro-partitionering.

Shuffles: de Duurste Operatie in Spark

Dit is het belangrijkste performance-concept in PySpark, en het verklaart waarom sommige operaties dramatisch trager zijn dan andere die er syntactisch vergelijkbaar uitzien. Een shuffle is het herverdelen van data over partities heen — nodig zodra een bewerking rijen met dezelfde sleutel bij elkaar moet brengen, ook al stonden ze oorspronkelijk op verschillende executors. Dit vereist het serialiseren, over het netwerk versturen, en op schijf schrijven van tussenresultaten — kostbaar op een manier die puur in-partition-bewerkingen niet zijn.

groupBy().agg(), join() (behalve broadcast joins, zie verderop), en distinct() triggeren allemaal een shuffle, omdat ze rijen met dezelfde key op dezelfde executor moeten samenbrengen om correct te kunnen aggregeren of matchen. filter(), select(), withColumn() shuffelen nooit — elke rij kan onafhankelijk, binnen zijn eigen partitie, worden verwerkt.

Wide versus Narrow Transformations

Dit onderscheid formaliseert precies het shuffle-verschil hierboven. Narrow transformations (filter, select, withColumn, map) hebben geen shuffle nodig — elke output-partitie hangt af van precies één input-partitie, wat volledig parallelle, onafhankelijke verwerking mogelijk maakt zonder enige communicatie tussen executors. Wide transformations (groupBy, join, distinct, orderBy) vereisen wél een shuffle — een output-partitie kan data nodig hebben uit meerdere, mogelijk alle input-partities.

# Narrow: elke rij onafhankelijk verwerkbaar, geen shuffle
df_clean = df.filter(col("amount") > 0).withColumn("amount_eur", col("amount") / 100)

# Wide: vereist een shuffle om alle rijen per categorie bij elkaar te brengen
df_summary = df_clean.groupBy("category").agg(F.sum("amount_eur").alias("total"))

De praktische consequentie: ketens van narrow transformations zijn vrijwel gratis om toe te voegen — Spark voert ze uit binnen dezelfde partitie zonder extra netwerk- of schijf-I/O. Elke wide transformation in een pipeline is een bewust punt waar je de kosten ervan zou moeten overwegen, vooral als er meerdere na elkaar voorkomen.

De Catalyst Optimizer: Waarom "Naïeve" Code Vaak Prima Is

Voordat een DataFrame-plan daadwerkelijk wordt uitgevoerd, gaat het door Spark's Catalyst optimizer, die het logische plan analyseert en herschrijft naar een efficiëntere fysieke uitvoeringsvorm — vergelijkbaar in geest met hoe een warehouse-queryplanner een SQL-query herschrijft vóór uitvoering. Concreet betekent dit onder andere predicate pushdown (een filter() na een join() in je code geschreven, wordt vaak vóór de join uitgevoerd als dat correct en sneller is) en kolomsnoei (alleen kolommen die daadwerkelijk gebruikt worden, worden ingelezen — relevant bij kolomgeoriënteerde formaten als Parquet uit Python for Data Engineers).

df.filter(col("category") == "electronics").explain()

.explain() toont het fysieke uitvoeringsplan zoals Catalyst het daadwerkelijk gaat uitvoeren — onmisbaar bij het debuggen van onverwacht trage queries, omdat het laat zien of een verwachte optimalisatie (pushdown, partition pruning) daadwerkelijk plaatsvond. Dit betekent niet dat queryvolgorde er nooit toe doet — Catalyst is goed, niet perfect — maar het verklaart waarom in de praktijk "logisch geschreven" code vaak al redelijk geoptimaliseerd wordt zonder dat je zelf elke stap handmatig hoeft te herordenen.

Partitionering in de Praktijk: Repartition en Coalesce

Het aantal partities beïnvloedt direct hoe goed werk parallelliseert, en Spark biedt twee manieren om dat aantal aan te passen, met een belangrijk verschil in kosten.

repartition(n) shuffelt data volledig opnieuw over n partities — kostbaar (het is zelf een wide transformation), maar noodzakelijk als je het aantal partities wilt vergroten, of data gelijkmatiger wilt herverdelen na een operatie die scheve partities achterliet.

coalesce(n) combineert bestaande partities zonder een volledige shuffle, en werkt alleen om het aantal partities te verkleinen — aanmerkelijk goedkoper dan repartition voor dat specifieke doel, bijvoorbeeld vlak vóór het wegschrijven van een resultaat om te veel kleine outputbestanden te vermijden.

df_repartitioned = df.repartition(200, "customer_id")  # herverdeel expliciet op sleutel
df_output = df_summary.coalesce(10)                     # verminder outputbestanden, goedkoop
df_output.write.format("delta").save("gold/summary")

Data skew — wanneer één partitiesleutel onevenredig veel rijen bevat (bijvoorbeeld één extreem grote klant tussen duizenden kleine) — is een van de meest voorkomende oorzaken van een Spark-job die "bijna klaar" blijft hangen: negenennegentig procent van de tasks is snel klaar, terwijl één executor blijft worstelen met een oneven grote partitie. Herpartitioneren op een andere, gelijkmatiger verdeelde sleutel, of het toevoegen van een "salt"-waarde aan de skewed key, zijn de gangbare oplossingen.

Broadcast Joins: een Shuffle Vermijden

Wanneer je een grote tabel joint met een kleine tabel (een fact table met een kleine dimension table, zie Star Schema & Dimensional Modeling), is een volledige shuffle-join onnodig kostbaar. Een broadcast join kopieert de kleine tabel volledig naar elke executor, zodat de join lokaal, zonder shuffle, kan plaatsvinden:

from pyspark.sql.functions import broadcast

df_result = large_orders_df.join(
    broadcast(small_dim_customers_df), on="customer_id"
)

Spark past dit vaak automatisch toe voor tabellen onder een configureerbare groottedrempel (spark.sql.autoBroadcastJoinThreshold), maar expliciet broadcast() gebruiken is verstandig zodra je zeker weet dat een tabel klein genoeg is en Spark's automatische schatting (gebaseerd op statistieken die niet altijd actueel zijn) dat zelf niet met zekerheid detecteert.

Cachen: Herhaalde Berekening Vermijden

Omdat lazy evaluation geen tussenresultaten onthoudt, wordt een DataFrame die meerdere keren wordt bevraagd (bijvoorbeeld eenmaal voor een count() en eenmaal voor een write()) standaard twee keer volledig herberekend. .cache() (of .persist() voor meer controle over opslaglocatie: geheugen, schijf, of beide) bewaart het resultaat na de eerste berekening expliciet:

df_expensive = df.join(other_df, "key").groupBy("category").agg(F.sum("amount"))
df_expensive.cache()

print(df_expensive.count())        # berekent en cachet nu
df_expensive.write.format("delta").save("gold/summary")   # hergebruikt de cache, geen herberekening

De vuistregel: cache een DataFrame zodra je hem meer dan één keer nodig hebt binnen dezelfde run, en ontcache (.unpersist()) expliciet zodra je klaar bent — cache die onnodig lang blijft hangen, verbruikt executor-geheugen dat andere taken nodig hebben.

Adaptive Query Execution: Optimaliseren Tijdens de Run

Catalyst optimaliseert vóór uitvoering, gebaseerd op statistieken die niet altijd accuraat zijn — vooral na complexe transformatieketens weet Spark vooraf vaak niet precies hoeveel rijen een tussenresultaat zal bevatten. Adaptive Query Execution (AQE), standaard ingeschakeld in moderne Spark-versies, past het uitvoeringsplan tijdens de run zelf aan, op basis van daadwerkelijk waargenomen tussenresultaten in plaats van vooraf geschatte statistieken.

Drie concrete dingen die AQE automatisch doet: het samenvoegen van te kleine shuffle-partities (die anders veel overhead per taak zouden veroorzaken relatief tot de hoeveelheid data), het splitsen of anders afhandelen van skewed partities (het probleem uit de vorige sectie, deels automatisch verzacht), en het dynamisch omzetten van een reguliere join naar een broadcast join zodra tijdens de run blijkt dat een van de twee tabellen kleiner is dan vooraf geschat. Dit betekent niet dat handmatige partitionering en expliciete broadcast joins overbodig zijn geworden — AQE werkt met wat het tijdens de run waarneemt, en een bewust ontworpen pipeline blijft voorspelbaarder dan volledig vertrouwen op runtime-aanpassingen — maar het verklaart waarom veel Spark-jobs tegenwoordig beter presteren dan de handmatige tuning-adviezen uit oudere documentatie zouden doen vermoeden.

spark.conf.set("spark.sql.adaptive.enabled", "true")               # standaard al aan
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")       # skew-afhandeling specifiek

Best Practices

• Ketens narrow transformations vrijelijk — ze zijn vrijwel gratis; wees terughoudender en bewuster met elke wide transformation (groupBy, join, distinct).
• Gebruik .explain() bij elke onverwacht trage query vóórdat je gokt naar een oplossing — het toont exact welke optimalisaties wel en niet zijn toegepast.
• Cache expliciet zodra een DataFrame meer dan eenmaal wordt gebruikt, en ontcache zodra je klaar bent.
• Gebruik broadcast joins bewust voor fact/dimension-achtige joins waar de kleine tabel ruim onder de automatische drempel valt.
• Gebruik coalesce in plaats van repartition wanneer je alleen het aantal partities wilt verkleinen — de kostenbesparing is direct en aanzienlijk.
• Laat AQE ingeschakeld staan (het is standaard) en behandel het als een vangnet bovenop, niet als vervanging van, bewuste partitionering en expliciete broadcast joins.

Veelgemaakte Fouten

• Meerdere acties op dezelfde onbecachte DataFrame aanroepen, waardoor dezelfde dure berekening onnodig meerdere keren wordt uitgevoerd.
• repartition gebruiken waar coalesce had volstaan, wat een onnodige volledige shuffle veroorzaakt voor iets dat goedkoper kon.
• Data skew negeren totdat een job structureel traag blijft hangen, in plaats van vroeg te herkennen dat één partitiesleutel de rest domineert.
• Een expliciete broadcast join forceren op een tabel die eigenlijk te groot is, wat elke executor met een onnodig zware kopie belast en geheugenproblemen kan veroorzaken.
• .cache() overal toepassen "voor de zekerheid", wat executor-geheugen verspilt aan tussenresultaten die maar één keer worden gebruikt en dus geen enkel voordeel van cachen hebben.

Performance Tips

• Filter en selecteer kolommen zo vroeg mogelijk in de keten — narrow transformations vóór een shuffle verkleinen de hoeveelheid data die geshuffled moet worden.
• Controleer partitieaantal na een zware filter — een sterk gefilterde DataFrame behoudt vaak het oorspronkelijke, nu te hoge partitieaantal, wat overhead per taak veroorzaakt op steeds kleinere hoeveelheden data per partitie.
• Gebruik kolomgeoriënteerde formaten (Parquet/Delta) als bron, nooit CSV voor zware Spark-workloads — Catalyst kan dan kolomsnoei en predicate pushdown toepassen, wat bij CSV niet mogelijk is.
• Monitor de Spark UI actief bij prestatieproblemen — het toont per-stage en per-task-duur, en identificeert direct welke specifieke taak (vaak door skew) veel langer duurt dan de rest.

Interviewvragen

"Wat is het verschil tussen een transformation en een action in Spark?" Let op: transformations zijn lui en bouwen alleen een uitvoeringsplan; actions triggeren de daadwerkelijke berekening — met een concreet voorbeeld van waarom een fout pas bij de action zichtbaar wordt.

"Wat is een shuffle, en waarom is het duur?" Let op: het herverdelen van data over partities heen zodat rijen met dezelfde sleutel samenkomen, wat serialisatie, netwerkverkeer en schijf-I/O vereist — wezenlijk duurder dan binnen-partitie-verwerking.

"Wat is het verschil tussen wide en narrow transformations?" Let op: narrow (filter, select) heeft geen shuffle nodig omdat elke output-partitie van precies één input-partitie afhangt; wide (groupBy, join) vereist wel een shuffle.

"Wanneer zou je een broadcast join gebruiken, en waarom is het sneller dan een reguliere join?" Let op: bij het joinen van een grote met een kleine tabel — de kleine tabel wordt naar elke executor gekopieerd, waardoor de join lokaal plaatsvindt zonder shuffle.

"Wat is het verschil tussen repartition en coalesce?" Let op: repartition shuffelt volledig en kan het aantal partities vergroten of verkleinen; coalesce combineert partities zonder volledige shuffle, maar alleen om te verkleinen.

"Wat is data skew, en hoe herken je het?" Let op: een ongelijke verdeling van data over partitiesleutels, herkenbaar doordat vrijwel alle tasks snel klaar zijn terwijl één of enkele executors veel langer blijven werken — zichtbaar in de Spark UI.

"Wat doet Adaptive Query Execution, en waarom is het nodig naast Catalyst's vooraf-optimalisatie?" Let op: Catalyst optimaliseert vóór uitvoering op basis van (soms onnauwkeurige) statistieken; AQE past het plan tijdens de run aan op basis van daadwerkelijk waargenomen tussenresultaten — bijvoorbeeld door een join dynamisch naar broadcast om te zetten of te kleine shuffle-partities samen te voegen.

Relevante Documentatie

• Apache Spark: RDD Programming Guide — Transformations and Actions — de officiële uitleg van lazy evaluation.
• Apache Spark: Performance Tuning — shuffles, broadcast joins en Catalyst in detail.
• Databricks: Understanding the Spark UI — hoe je stages, tasks en skew in de praktijk identificeert.

Samenvatting

PySpark's gedrag wordt volledig verklaard door een klein aantal onderliggende mechanismen: lazy evaluation (transformations bouwen een plan, actions voeren het uit), partitionering over executors als de fysieke basis van parallellisatie, en shuffles als de duurste operatie — het onderscheid tussen wide en narrow transformations formaliseert precies wanneer die shuffle optreedt. De Catalyst optimizer herschrijft plannen automatisch voor efficiëntie, maar .explain() blijft onmisbaar om te verifiëren dat die optimalisatie daadwerkelijk plaatsvond. Repartition/coalesce, broadcast joins en expliciet cachen zijn de directe hendels om controle te nemen over kosten die anders impliciet en onzichtbaar blijven. Met deze mechaniek begrepen, is het volgende hoofdstuk de logische vervolgstap: Delta Lake, de tabellaag die ACID-garanties toevoegt aan precies de gedistribueerde bestanden die Spark hier verwerkt.