Spark & PySpark

Was macht Spark & PySpark?

Nahezu jeder langsame Spark-Job ist aus einem von drei Gründen langsam: ein vermeidbarer Shuffle, ein schiefer Key (Skew) oder Zeilen, die Python einzeln durchlaufen. Der teure Fehler ist, die Clustergröße zu erhöhen, bevor man den Plan liest – also mehr Executoren zu bezahlen, um denselben verschwenderischen DAG schneller auszuführen.

Vorgehensweise

Schritt 1: Eingaben sammeln

1. Den Job-Code und bei einem langsamen Job die Spark-UI-Ansicht der langsamen Stufe (Task-Dauer-Verteilung, Shuffle-Lese-/Schreibgrößen) sowie die Ausgabe von df.explain(True).
2. Eingabedatengröße und -format, sowie die Größen beider Seiten jedes Joins.
3. Cluster-Form: Anzahl Executoren, Kerne und Speicher.
4. Ob es sich um eine Neuentwicklung oder eine Optimierung handelt – bei Optimierung zuerst aus dem Plan diagnostizieren, bevor der Code geändert wird.

Schritt 2: Mit einer vernünftigen Session starten

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (
    SparkSession.builder
    .appName("etl")
    .config("spark.sql.shuffle.partitions", "200")
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)

Adaptive Query Execution (AQE) aktivieren; es fasst Shuffle-Partitionen zusammen und wechselt Join-Strategien zur Laufzeit. Dann spark.sql.shuffle.partitions bewusst setzen: Ziel sind 100-200 MB pro Shuffle-Partition, also Partitionen ≈ Shuffle-Datengröße ÷ 128 MB, und mindestens das 2-3-fache der gesamten Executor-Kerne, damit kein Kern untätig bleibt.

Schritt 3: Im DataFrame-API bleiben und Python vermeiden

DataFrame-Operationen durchlaufen Catalyst und Tungsten, was Abfrageoptimierung und Off-Heap-Speicher ermöglicht. Nur dann auf RDDs zurückgreifen, wenn kein DataFrame-Primitiv existiert. Python-UDFs serialisieren jede Zeile zum Python-Worker und zurück und brechen die Codegenerierung. Reihenfolge der Präferenz:

1. Eingebaute pyspark.sql.functions (z.B. F.regexp_extract, F.when).
2. Pandas (vektorisierte) UDFs, wenn Python unvermeidbar ist.
3. Skalare Python-UDFs nur als letztes Mittel.

from pyspark.sql.functions import pandas_udf

@pandas_udf("double")
def normalize(s):
    return (s - s.mean()) / s.std()

Schritt 4: Join-Strategie nach Größe wählen

• Die kleinere Seite broadcasten, wenn sie in den Speicher passt. Spark broadcastet automatisch unterhalb von spark.sql.autoBroadcastJoinThreshold (Standard 10 MB); Tabellen bis zu einigen hundert MB explizit broadcasten, wenn der Executor-Speicher es erlaubt, und ~1 GB als praktische Obergrenze behandeln – Broadcasten kopiert die Tabelle auf jeden Executor und führt bei Fehleinschätzung zu OOM des Jobs.

from pyspark.sql.functions import broadcast
result = large.join(broadcast(small), "key")

• Vor dem Join filtern und Spalten projizieren, um den Shuffle zu verkleinern; jede weggelassene Spalte und Zeile sind gesparte Shuffle-Bytes.
• Zwei große Tabellen → den Sort-Merge-Join akzeptieren, aber zuerst auf Skew prüfen (Schritt 5).

Schritt 5: Skew diagnostizieren und beheben

Skew-Symptome: Die Stufe ist fertig, bis auf ein oder zwei Tasks; die maximale Task-Dauer übersteigt das ~3-5-fache des Medians; eine Shuffle-Read-Partition ist in der Spark-UI weit größer als die anderen. AQEs Skew-Join-Handling teilt eine Partition, wenn sie sowohl über das 5-fache der medianen Partitionsgröße als auch über 256 MB (die Standardwerte) liegt – überprüfen, ob es im Plan ausgelöst wurde. Wenn AQE nicht helfen kann (z.B. bei schiefer Aggregation), die heißen Keys salzen: einen zufälligen Suffix 0..N an den Key der großen Seite anhängen, die kleine Seite über alle N Suffixe aufblähen, joinen und dann das Salz entfernen. N ≈ den Skew-Faktor wählen (ein Key, der 20x den Median beträgt, erhält N=20).

Schritt 6: Partitionierung und Ausgabelayout steuern

• repartition(n, col) führt einen vollständigen Shuffle durch, um Daten auszugleichen; vor breiten Schreibvorgängen verwenden.
• coalesce(n) reduziert Partitionen ohne Shuffle; verwenden, um winzige Ausgabedateien zu vermeiden.
• Ziel: Ausgabedateien von 128 MB bis 1 GB; tausende kleiner Dateien bestrafen jeden nachgelagerten Leser.
• Ausgabe nur nach niedrigkardinalen Spalten partitionieren (Datum, Region – Dutzende bis Hunderte von Werten, niemals IDs):

df.write.partitionBy("dt").mode("overwrite").parquet("/data/out")

Schritt 7: Nur cachen, was wiederverwendet wird

Nur cachen, wenn ein DataFrame mehrfach im DAG verwendet wird; materialisieren und nach Gebrauch unpersistieren:

df.cache()
df.count()  # materialisieren
# ... df wiederverwenden ...
df.unpersist()

Das Cachen eines nur einmal verwendeten DataFrames verschwendet Speicher und kann Daten verdrängen, die wichtig waren.

Schritt 8: Mit dem Plan verifizieren

df.explain(True) erneut ausführen und bestätigen, dass die beabsichtigte Strategie erscheint (BroadcastHashJoin, pushed filters, kein Exchange, wo eines entfernt wurde). Spaltenorientierte Formate (Parquet, ORC) lesen, damit Predicate Pushdown funktioniert; niemals collect() auf große Daten anwenden, um sie zum Treiber zu bringen; und bei Verkettungen F.col-Referenzen gegenüber String-Spalten bevorzugen.

Ausgearbeitetes Beispiel: Schlechter vs. guter Join

Schlecht – eine 2 TB große Ereignistabelle gejoint mit einer 200 MB Dimension, volle Breite, Python-UDF für das Parsen:

result = (events.join(dims, "store_id")          # sort-merge: mischt alle 2 TB
    .withColumn("region", parse_region_udf("meta"))  # Python Zeile für Zeile
    .filter(F.col("dt") == "2024-06-01"))            # filter NACHDEM gejoint wurde

Der Plan zeigt einen Exchange auf beiden Seiten (2 TB gemischt), und die UDF blockiert die Codegenerierung.

Gut – zuerst filtern und projizieren, die Dimension broadcasten, eine eingebaute Funktion verwenden:

result = (events
    .filter(F.col("dt") == "2024-06-01")              # zuerst auf ~30 GB reduzieren
    .select("store_id", "amount", "meta")
    .join(broadcast(dims.select("store_id", "region_code")), "store_id")
    .withColumn("region", F.regexp_extract("meta", r"region=(\w+)", 1)))

Der Plan zeigt jetzt BroadcastHashJoin ohne großen Exchange: ~30 GB gescannt, kein Shuffle der großen Tabelle, keine Python-Grenze.

Ergebnis

Den überarbeiteten Job-Code sowie eine Optimierungsnotiz liefern, die enthält: die Join-Strategie pro Join mit Größenbeleg, die Einstellung der Shuffle-Partitionen mit der zugrundeliegenden Berechnung, jeden gefundenen Skew (maximales vs. medianes Task-Verhältnis) und die angewandte Behebung, Caching-Entscheidungen, Erwartungen an die Ausgabedateigrößen sowie die Vorher/Nachher-explain-Unterschiede bei einer Optimierung.

Nicht tun

• Eine Tabelle NICHT broadcasten, deren Größe nicht bekannt ist – ein falsch eingeschätzter Broadcast führt bei allen Executoren gleichzeitig zu OOM.
• KEINE Python-UDF schreiben, bevor die pyspark.sql.functions überprüft wurden; fast alle String-, Datums- und Bedingungslogiken sind eingebaut.
• NICHT collect() oder toPandas() auf einen großen DataFrame anwenden; das leitet die Daten des gesamten Clusters durch einen einzigen Treiber.
• Ausgabe NICHT nach einer hochkardinalen Spalte partitionieren – Millionen von Verzeichnissen mit jeweils einer Zeile.
• NICHT einfach Executoren auf einen Job werfen, dessen langsame Stufe ein einzelner schiefer Task ist; mehr Kerne können eine Partition nicht aufteilen.
• NICHT reflexartig cachen; nur DataFrames cachen, die nachgelagert wiederverwendet werden, und sie wieder unpersistieren.

Qualitätsmaßstab

Ein fertiger Job gilt nur dann als bestanden, wenn: jeder Join seine Strategie und den Größenbeleg dafür benennt; spark.sql.shuffle.partitions aus der Datengröße abgeleitet ist und nicht auf einem Gewohnheitswert belassen wurde; die Spark-UI in Shuffle-Stufen keinen Task über ~3x der medianen Dauer zeigt (oder der Skew erklärt und behoben ist); keine skalare Python-UDF überlebt, wo eine eingebaute Funktion existiert; Ausgabedateien im Bereich von 128 MB bis 1 GB liegen; und der endgültige explain den beabsichtigten Plan bestätigt.