Warum dein Spark-Job weiterhin mit OOM-Fehlern abstürzt, selbst nachdem du Amazon EMR skalierst

Amazon EMR konfiguriert spark.executor.cores und spark.executor.memory automatisch anhand der im Cluster verwendeten EC2-Instance-Typen. AWS dokumentiert das ausdrücklich: Beide Werte werden aus den Core- und Task-Instances abgeleitet.

Das ist praktisch. EMR kennt die verfügbaren CPU-Kerne und weiß, wie viel Speicher YARN über yarn.nodemanager.resource.memory-mb bereitstellt. Daraus werden die Executors so dimensioniert, dass möglichst wenig CPU oder RAM ungenutzt bleibt.

Was EMR jedoch nicht weiß: ob deine Joins, Aggregationen oder stark verzerrten Partitionen tatsächlich in diesen Heap passen. AWS weist in den eigenen Performance-Empfehlungen darauf hin, dass manche Workloads größere Executor-JVMs benötigen, während andere mit kleineren und zahlreicheren Executors besser laufen. EMR löst ein Ressourcenverteilungsproblem. Die Dimensionierung für den eigentlichen Workload bleibt deine Aufgabe.

Homogene Fleet: optimal gepackte Maschinen, kleine Heaps

Nehmen wir eine Fleet aus c5.12xlarge-Instances mit 48 vCPUs und 96 GiB RAM. Im Troubleshooting-Beispiel von AWS stehen YARN 90.112 MiB zur Verfügung und EMR landet bei:

spark.executor.cores = 3
spark.executor.memory = 4743 MiB

Zusätzlich verwendet EMR spark.yarn.executor.memoryOverheadFactor = 0.1875 (seit EMR 5.22), wodurch jeder YARN-Container ungefähr 5.632 MiB benötigt. Sowohl anhand der CPU-Kerne als auch des Speichers passen damit exakt 16 Executors auf eine Instance. Perfekt gepackt.

Dann kommt eine Stage mit großen Joins, und einige Tasks benötigen mehr als ungefähr 4,7 GiB Heap.

Der gesamte Cluster kann dabei mehrere Terabyte RAM besitzen. Das hilft nicht.

Ein Spark-Task läuft innerhalb einer einzelnen Executor-JVM. Freier Speicher auf einer anderen Node macht diesen Executor nicht größer. Ihm geht der Speicher aus und er läuft in einen OOM-Fehler.

Mehr Nodes machen den Executor nicht größer

Die Fehler sehen typischerweise so aus:

ExecutorLostFailure
Container killed by YARN for exceeding memory limits
java.lang.OutOfMemoryError

Der erste Impuls lautet oft: „Wir brauchen mehr Speicher.“

Also wird die Fleet von 10 auf 20 Worker vergrößert.

Gleicher Instance-Typ, gleiche Executor-Konfiguration: Jetzt gibt es 320 statt 160 Executors, aber jeder einzelne hat weiterhin nur ungefähr 4,7 GiB Heap.

Du hast den Task, der nicht in den verfügbaren Speicher passt, nicht repariert.

Du hast lediglich mehr Kopien derselben Einschränkung gekauft.

Scale-out kann Timing, Parallelität und Shuffle-Druck ausreichend verändern, sodass der Job manchmal doch erfolgreich durchläuft. Das bedeutet jedoch nicht, dass du einen guten Kosten-Performance-Punkt erreicht hast.

Unter Umständen zahlst du für die doppelte Menge an EC2-Ressourcen, nur um zu vermeiden, zwei Spark-Konfigurationswerte anzupassen.

Die entscheidende Frage lautet nicht, ob der Cluster insgesamt genug Speicher besitzt.

Die Frage lautet, ob dieser Executor genug Speicher für die Arbeit besitzt, die ihm zugewiesen wurde.

Gemischte Instance-Familien

Komplizierter wird es, wenn eine Fleet Instance-Typen mit sehr unterschiedlichen CPU-zu-Speicher-Verhältnissen kombiniert. Laut den EMR Task Configuration Tables:

m5.4xlarge  16 vCPU  57.344 MiB YARN
r5.4xlarge  16 vCPU  122.880 MiB YARN

Gleiche Anzahl an CPU-Kernen, aber mehr als doppelt so viel Speicher.

Im Performance Guide von AWS wird das mit c5-, m5- und r5.4xlarge noch deutlicher: Alle haben 16 vCPUs, stellen YARN aber ungefähr 24, 57 beziehungsweise 123 GiB Speicher zur Verfügung.

Mit EMR 5.32 und 6.5 wurden heterogene Executors standardmäßig aktiviert (spark.yarn.heterogeneousExecutors.enabled). Die Executor-Größe kann sich dadurch nach dem jeweiligen Instance-Typ richten, statt ein einziges Profil für die gesamte Fleet zu verwenden. Über spark.executor.maxMemory und spark.executor.maxCores lassen sich zudem Obergrenzen festlegen.

Das verbessert die Ressourcenverteilung.

Es macht den Workload aber nicht homogen.

Eine besonders große Partition kann auf einem kleinen Executor landen, dort einen OOM-Fehler verursachen, erneut ausgeführt werden, wieder auf einem kleinen Executor landen und schließlich auf einer R-Instance erfolgreich durchlaufen.

Der Job meldet am Ende SUCCESS.

Bezahlt hast du trotzdem für alle fehlgeschlagenen Versuche und für jede andere Node, die während der erneuten Berechnungen weiterläuft.

Worker, die Tasks annehmen, die sie nicht erfolgreich abschließen können, verschwenden nicht nur ihre eigene Rechenzeit. Sie verlängern die Laufzeit des gesamten Clusters.

Eine günstigere Instance, die mitten im Task abstürzt, kann deshalb am Ende teurer sein als eine R-Instance, die denselben Task beim ersten Versuch erfolgreich abschließt.

Begrenze die Fleet auf das, was der Workload tatsächlich benötigt

Instance Fleets sind hervorragend für Spot-Diversifizierung und bessere Kapazitätsverfügbarkeit geeignet.

Sie sind aber keine Einladung, jeden Instance-Typ zuzulassen, auf dem YARN theoretisch einen Container starten kann.

Die Flexibility-Empfehlungen von AWS schlagen die richtige Reihenfolge vor: Bestimme zuerst den kleinsten Executor, den dein Workload benötigt, und erlaube anschließend nur Instance-Typen, die diesen Executor sinnvoll ausführen können.

Wenn du beispielsweise mindestens 2 vCPUs und 8 GiB Speicher pro Executor benötigst, bildet m5.xlarge die Untergrenze für diese Fleet.

Wenn eine Maschine keinen sinnvollen Executor für diesen Job bereitstellen kann, ist sie keine zusätzliche Kapazität.

Sie ist lediglich ein Ort, an dem Tasks sterben.

Behandle die Spark-Konfiguration als Teil der Anwendung

Die EMR-Defaults verändern sich.

Dynamic Allocation ist seit EMR 4.4 standardmäßig aktiviert. Der Overhead-Faktor von 18,75 % kam mit EMR 5.22. Heterogene Executors folgten mit EMR 5.32 beziehungsweise 6.5.

Was auf einer bestimmten EMR-Version und Instance-Familie problemlos funktioniert hat, kann nach einem Wechsel einer dieser Komponenten plötzlich ungeeignet sein.

Executor Memory, CPU-Kerne, Task-Größe, Shuffle, Garbage Collection, Data Skew und der verwendete EC2-Mix müssen gemeinsam betrachtet werden.

Meistens besteht die Lösung nicht darin, ein weiteres Dutzend Nodes hinzuzufügen.

Oft reichen einige gezielte Entscheidungen bei Executor-Konfiguration und Instance Fleet.

Für jemanden mit tiefgehender Erfahrung mit Spark, YARN und EMR lässt sich ein deutlich besserer Kosten-Performance-Punkt häufig innerhalb weniger Stunden finden, ohne die gesamte Plattform neu aufzubauen.

Das ist günstig im Vergleich dazu, denselben Ressourcenverbrauch bei jedem Produktionslauf erneut zu bezahlen.

Brauchst du Unterstützung?

Wenn Spark auf EMR teuer, instabil oder langsamer ist, als es sein sollte, kann ich helfen.

Während meiner Zeit bei AWS habe ich viel Zeit mit tiefgehenden Spark- und EMR-Analysen verbracht. Häufig konnten dadurch Laufzeiten um 30 % oder mehr reduziert werden, bei entsprechend ähnlichen Kosteneinsparungen.

Bei regelmäßig ausgeführten Jobs können daraus schnell mehrere Tausend Dollar pro Monat werden: weniger EC2-Stunden und weniger Rechenleistung, die niemals produktive Arbeit geleistet hat.

Hinweis: Dieser Artikel wurde mithilfe von KI aus dem englischen Original ins Deutsche übersetzt.