Was passiert eigentlich, wenn Sie einen Spark-Job ausführen?

Einführung

Mit zunehmender Größe von Datensätzen kann ihre effiziente Verarbeitung auf einem einzelnen Rechner aufgrund begrenzter CPU-Kapazität, begrenzten Arbeitsspeichers und begrenzter Speicherbandbreite zur Herausforderung werden.

Apache Spark begegnet diesen Herausforderungen mit einer verteilten Datenverarbeitungs-Engine, die Arbeitslasten in Partitionen aufteilen und Aufgaben parallel über CPU-Kerne und mehrere Rechner hinweg ausführen kann.

Bei Spark geht es jedoch nicht einfach darum, weitere Rechner hinzuzufügen.

Für den Entwurf effizienter und zuverlässiger Datenpipelines ist es entscheidend zu verstehen, wie die einzelnen Komponenten zusammenarbeiten: Treiber, Executors, RDDs, DataFrames, Partitionen, Jobs, Stages und Tasks.

In diesem Artikel erläutere ich anhand eines praxisnahen Beispiels zur Aggregation von E-Commerce-Umsätzen die Architektur von Spark, seine Ausführungsmechanismen und wichtige Aspekte der Performance.

1. Warum Apache Spark? – Parallele Datenverarbeitung

Apache Spark ist eine verteilte Rechen-Engine für die Verarbeitung großer Datenmengen und Data Engineering.

Spark verarbeitet Big Data mithilfe paralleler Datenverarbeitung:

Abbildung A. Überblick über die verteilte Apache-Spark-Clusterarchitektur (Cluster, Treiber- und Worker-Knoten) im Vergleich zur Ausführung auf einem einzelnen Knoten (erstellt von Kuriko IWAI)

Im Gegensatz zur Verarbeitung auf einem einzelnen Knoten (rechts, Abbildung A) teilt Spark die Arbeit in Tasks auf und führt sie – sofern verfügbar – auf mehreren Knoten und über die Ressourcen verschiedener Rechner hinweg aus.

„Verteilt“ bedeutet hier nicht, dass mehrere Rechner erforderlich sind. Spark kann im lokalen Modus auf einem einzelnen Rechner laufen. Auch dann lassen sich Daten partitionieren und parallel verarbeiten.

2. Die Kernarchitektur von Apache Spark

Spark besteht aus mehreren zentralen Komponenten:

• Spark-Treiber (Treiber): Der zentrale Koordinationsprozess einer Spark-Anwendung.
• Cluster-Manager: Ein externer Dienst, der für die Zuweisung von Hardwareressourcen wie CPU-Kernen und Arbeitsspeicher zuständig ist.
• Worker-Knoten (Worker): Ein physischer oder virtueller Rechner beziehungsweise Rechenknoten im Cluster.
• Executors: Verteilte Worker-Prozesse, die auf Worker-Knoten gestartet werden.
• Java Virtual Machine (JVM): Die Laufzeitumgebung, die zum Ausführen des Spark-Bytecodes erforderlich ist (da Spark in Scala/Java geschrieben ist).

Zunächst koordiniert der Treiber eine Spark-Anwendung und wandelt sie in einen gerichteten azyklischen Graphen (DAG) aus physischen Ausführungsaufgaben um. Anschließend verteilt er diese auf die Executors.

Anschließend weist der Cluster-Manager Ressourcen gemäß dem Ressourcenverwaltungssystem der Bereitstellung zu.

Worker-Knoten stellen den Executors Rechenressourcen bereit. Diese führen Tasks aus und halten während ihrer Ausführung Daten vor.

In einigen verwalteten Umgebungen wie Databricks wird pro Worker ein Executor eingesetzt; dies ist jedoch keine allgemeingültige Spark-Regel. Je nach Clusterkonfiguration kann ein Worker mehrere Executors beherbergen.

2-1. E-Commerce-Anwendungsfall – Struktur der Eingabedaten

Stellen wir uns ein kleines E-Commerce-Unternehmen vor.

Es verfügt über 10.000 Verkaufsdatensätze, die jeweils eine Bestandseinheit (SKU) und die verkaufte Menge enthalten. Es seien drei mögliche SKUs angenommen: A, B, und C.

Tabelle 1. Beispiel für die Aufschlüsselung von E-Commerce-Verkaufsdatensätzen und die erwarteten Aggregationsergebnisse.

Unser Ziel ist es, die Gesamtzahl der gültig verkauften Einheiten pro SKU zu berechnen:

1. Datensätze herausfiltern, deren qty <= 0, dann
2. Datensätze nach SKU aggregieren.

Nehmen wir für den vollständigen Beispieldatensatz an, dass nach dem Filtern 9.800 der 10.000 Datensätze übrig bleiben. Die von Spark erwarteten Endergebnisse sind:

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

2-2. Spark-Cluster

Für unseren Anwendungsfall lege ich ein vereinfachtes Cluster fest.

Worker-Knoten (Worker)

Ich lege drei Worker-Knoten fest, die jeweils über 8 vCPU und 32 GiB RAM verfügen:

Abbildung B. Ressourcenzuweisung auf die Worker-Knoten (erstellt von Kuriko IWAI)

Insgesamt stehen dem System 24 vCPU zur Verfügung.

Spark-Executors und CPU-/Speicherzuweisung

Nehmen wir an, ich sichere mir:

• 7 Executors.
• 2 vCPU-Kerne pro Executor.
• 3 GiB JVM-Heap-Speicher pro Executor.

Die Spark-Anwendung fordert den Cluster-Manager auf, diesen 7 Executors Ressourcen zuzuweisen.

Anschließend legt der Cluster-Manager die Zuweisung von CPUs und Speicher anhand der Ressourcennutzung der einzelnen Worker fest.

Zum Beispiel entscheidet er, 14 Task-Slots zu erstellen – 2 vCPU-Kerne pro Executor × 7 Executors:

Abbildung C-1. Zuweisung von vCPU-Kernen, Speicher und Executors auf die Worker bei unausgewogener Ressourcennutzung auf Worker A

Beachten Sie, dass die dem Executor zugewiesenen 3 GiB nicht dessen gesamtem Speicher entsprechen, da zusätzlicher Speicherbedarf für Worker und andere Ausführungen berücksichtigt werden muss.

Angenommen, eine einzelne Aufgabe benötigt 1 vCPU (spark.task.cpus = 1), weist jeder Worker seine vCPUs wie folgt zu:

Abbildung C-2. Parallele Zuweisung von vCPUs auf die Worker

Daran werden zwei wichtige Punkte deutlich:

1. Jeder Worker hat eine eigene CPU-Auslastung (in Abbildung C-2 beträgt die Auslastung von Worker A 75 %, während Worker B und C bei 50 % bleiben).
2. Obwohl das System über 24 vCPU-Kerne verfügt (8 vCPU pro Worker × 3 Worker), kann es nur 14 vCPU-Kerne nutzen; die übrigen 10 bleiben für diese Aufgaben ungenutzt.

2-3. Die Java Virtual Machine (JVM) hinter Spark

Apache Spark basiert auf der virtuellen Java-Maschine (JVM).

Die JVM ist eine Laufzeitumgebung, die zum Ausführen des Spark-Bytecodes benötigt wird.

Da Spark in Scala/Java geschrieben ist:

• Bei der Verwendung von Scala oder Java in einer Anwendung: Der Code wird nativ innerhalb der JVM ausgeführt.
• Bei der Verwendung von PySpark (Python): Ein Python-Prozess führt den Code aus, kommuniziert jedoch über eine Brücke wie Py4J mit einem JVM-basierten Treiber. Die Spark-Executors auf den Worker-Knoten laufen weiterhin innerhalb einer JVM, die dann Python-Worker zur Ausführung von Python-Code und benutzerdefinierten Python-Funktionen (Python-UDFs) verwaltet.

In beiden Fällen verwaltet die JVM die Ausführungsthreads, die Speicherzuweisung und die Garbage Collection sowohl für den Treiber als auch für die Executors.

JVMs von Treiber & Executors in einer PySpark-Anwendung

In einer PySpark-Anwendung kommuniziert der Python-Treiber mit dem in der JVM laufenden Spark-Treiber;

• Python-Treiber: Hier wird der PySpark-Code auf Anwendungsebene geschrieben und ausgeführt.
• Treiber-JVM: Koordiniert die Spark-Ausführung und die Planung.
• Executor-JVM: Führt Spark-Aufgaben aus und verwaltet Ausführungsressourcen. Wenn Aufgaben Python-Code ausführen müssen, kommen Python-Worker-Prozesse zum Einsatz.
• Client-JVM (Client-Prozess oder Übermittlungsrechner): JVM, die auf dem lokalen Rechner oder einem Edge-/Gateway-Knoten läuft.

Wie in der Abbildung dargestellt, verbindet sich die Client-JVM zunächst mit der Treiber-JVM, die anschließend die Executor-JVMs koordiniert:

Abbildung D. Gesamtbeziehung zwischen Client-, Treiber- und Executor-JVMs

Von hier aus beherbergt die Treiber-JVM den Spark-Treiber, der die Anwendung verwaltet, während die Executor-JVMs die einzelnen Executors beherbergen, die für die Ausführung der Aufgaben zuständig sind.

Jeder JVM-Treiber verfügt über einen JVM-Heap, um Datenverarbeitung, Caching und Benutzerobjekte effizient zu verwalten.

2-4. Resilient Distributed Datasets (RDDs) im Detail

Ein Resilient Distributed Dataset (RDD) ist eine unveränderliche, partitionierte Sammlung von Elementen, die Spark parallel verarbeiten kann.

RDDs halten Abhängigkeiten zwischen Transformationen fest. Dadurch kann Spark in vielen Fehlerszenarien verlorene Partitionen neu berechnen.

Der Name bringt drei Eigenschaften zum Ausdruck:

• Resilient (fehlertolerant): Spark speichert die Abstammung – also, von welchem vorherigen RDD ein RDD abhängt – und kann fehlende Partitionen häufig rekonstruieren, anstatt Replikate jedes Zwischenergebnisses vorzuhalten.
• Distributed (verteilt): Der Datensatz wird als Partitionen dargestellt, die Spark auf verschiedenen Executors verarbeiten kann. Mehrere physische Rechner sind nicht erforderlich; auch der lokale Modus unterstützt RDDs.
• Dataset (Datensatz): Ein RDD kann Schlüssel-Wert-Paare, Ganzzahlen, Zeichenfolgen, Tupel und andere unterstützte Elemente enthalten – nicht nur relationale Zeilen. Die RDD-APIs umfassen unter anderem Operationen wie map, filter, reduceByKey, join, und aggregateByKey.

Ein RDD ist eine Abstraktion für eine logische Datensammlung und Berechnung und keine Festlegung, dass bestimmte Partitionen dauerhaft bestimmten Workern zugewiesen sind.

Partitionen

Eine Partition ist ein logischer Datenblock und entspricht in der Regel einer Aufgabe innerhalb einer Stage.

Nehmen wir an, wir erstellen 100 Partitionen (diese 100 anfänglichen Partitionen dienen ausdrücklich der Veranschaulichung).

Bei 10.000 Quellelementen ergeben sich ungefähr 100 Elemente pro Partition, wenn die Eingabedaten annähernd gleichmäßig verteilt sind:

Abbildung E. Parallele Zuweisung auf 100 anfängliche Datensplits

3. RDD-API vs. DataFrame-API: interne Catalyst-Optimierung

Das ursprüngliche RDD-Beispiel hilft dabei, Abstammung, Partitionen und Shuffle zu verstehen.

Bei typischen strukturierten ETL-Prozessen ist die DataFrame-API jedoch der bevorzugte Ausgangspunkt, da sie Spalten, Schemainformationen, SQL-ähnliche Ausdrücke und für den Optimierer sichtbare Operationen bereitstellt.

Das folgende Diagramm veranschaulicht den Zusammenhang:

Abbildung F. Zusammenhang zwischen den Graphkonzepten von Spark: RDD-Abstammung, DataFrame und Spark-Ausführungs-DAG

So wie ein RDD seine Abstammung verwendet, um die Transformationen zu definieren, stellt der Abfrageplan eines DataFrame dessen strukturierte Abfrage dar.

Spark erstellt aus Abhängigkeiten und physischen Operatoren die auszuführenden Arbeitsschritte.

Bei der Ausführung von DataFrames können intern RDD-basierte Mechanismen zum Einsatz kommen. Das bedeutet jedoch nicht, dass Nutzer beim Schreiben von DataFrame-Code die RDD-Abstammung manuell erstellen müssen.

3-1. DataFrame-Abfrageplan

Ein DataFrame-Abfrageplan ist die Abfolge logischer und physischer Schritte, die eine Ausführungs-Engine erzeugt (violetter Kasten in Abbildung L).

Eine DataFrame-Abfrage durchläuft mehrere Darstellungsformen:

1. Logischer Plan: Beschreibung der angeforderten Operationen (Scannen, Filtern, Gruppieren, Aggregieren).
2. Analysierter und optimierter logischer Plan: Spark löst Referenzen auf und wendet geeignete logische Optimierungen an, etwa Spaltenbeschneidung oder Prädikat-Pushdown.
3. Physischer Plan: Spark wählt ausführbare Operatoren aus, z. B. Scans, Teil- und Endaggregation, Exchanges, Joins und Sortierungen.
4. Ausführung: Der physische Plan wird mithilfe der Ausführungsmechanismen von Spark ausgeführt; letztlich kommen dabei Aufgaben auf Executoren zum Einsatz.

In unserem Anwendungsfall kann der physische Plan die folgenden Schritte umfassen:

• Partielle Hash-Aggregation,
• Exchange nach SKU und
• Abschließende Hash-Aggregation.

3-2. Vergleich von Aktionen: RDD-API vs. DataFrame-API

Vergleichen wir die RDD-API und die DataFrame-API am Beispiel der Aggregation nach SKU.

RDD-API

Die Elemente sind (sku, qty) Paare. Unsere Aggregationslogik in Python ist explizit, und Spark folgt den RDD-Abhängigkeiten:

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

Nehmen wir an, sales_df hat die Spalten sku (Zeichenkette) und qty (numerisch):

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)]

Beide Beispiele lösen hier dasselbe Geschäftsproblem:

Ungültige Verkaufsdatensätze herausfiltern und gültige Mengen nach SKU aggregieren.

Und in beiden Fällen collect() löst die Ausführung in Spark aus und gibt alle resultierenden Datensätze an den Driver zurück.

Der entscheidende Unterschied liegt hier in der Darstellung der Ergebnisse:

• RDD-API: Gibt Python-Elemente zurück, etwa (sku, total_qty) Tupel.
• DataFrame-API: Gibt strukturierte Row Objekte mit benannten Feldern zurück, etwa sku und total_qty.

Hier ein Gesamtvergleich der RDD- und DataFrame-APIs:

Tabelle 2. RDD-API im Vergleich zur DataFrame-API – Vergleich der Funktionen und Abstraktion: (Erstellt von Kuriko IWAI)

4. Zusammenfassung

Beim Spark-Tuning geht es nicht darum, die Anzahl der Partitionen oder die CPU-Auslastung für sich genommen zu maximieren.

Ziel ist es, die Ausführungskosten und die Latenz über den gesamten Ablauf hinweg zu senken, ohne Instabilität zu verursachen.

In diesem Artikel lernen wir die grundlegenden Zusammenhänge in seinem System kennen:

• Ressourcen: Worker beherbergen Executor; Executor führen Tasks aus; die konfigurierte CPU- und Speicherkapazität begrenzt die Parallelität.
• RDD-Programmierung: Transformationen erzeugen unveränderliche RDD-Definitionen und Lineage; Aktionen lösen die Berechnung aus.
• DataFrame-Programmierung: Transformationen erzeugen logische Abfrageoperationen; die Optimierung wählt ausführbare physische Operatoren aus; Aktionen lösen die Berechnung aus.
• Ausführung: Spark organisiert die Arbeit in Jobs, Stages und Tasks, und Tasks verarbeiten Partitionen auf Executors. Shuffles und andere Abhängigkeiten schaffen Koordinationsgrenzen zwischen den Stages.

Zu verstehen, wie Spark verteilte Arbeitslasten ausführt, ist erst der Anfang.

Sie können einen siebenschrittigen Berechnungsablauf mit der RDD-API kennenlernen.