Konfiguration von Spark auf YARN

Hallo Habr! Gestern beim Meetup über Apache Spark, das von den Leuten von Rambler&Co organisiert wurde, gab es viele Fragen der Teilnehmer zur Konfiguration dieses Werkzeugs. Wir haben beschlossen, unser Wissen darüber zu teilen. Das Thema ist nicht einfach — deshalb laden wir auch ein, Erfahrungen in den Kommentaren zu teilen; vielleicht verstehen wir auch etwas nicht richtig oder nutzen es falsch.

Eine kleine Einführung — wie wir Spark nutzen. Wir haben ein dreimonatiges Programm „Spezialist für Big Data“, und im gesamten zweiten Modul arbeiten unsere Teilnehmer mit diesem Tool. Folglich liegt es in unserer Verantwortung als Organisatoren, einen Cluster für diesen Anwendungsfall vorzubereiten.

Das Besondere an unserer Nutzung ist, dass die Anzahl der Personen, die gleichzeitig an Spark arbeiten, gleich der gesamten Gruppe sein kann. Zum Beispiel während eines Seminars, wenn alle gleichzeitig etwas ausprobieren und dem Lehrer nachahmen. Das können manchmal 40 Personen sein. Wahrscheinlich gibt es nicht viele Unternehmen weltweit, die mit einem solchen Nutzungsszenario konfrontiert sind.

Im Folgenden werde ich erklären, wie und warum wir bestimmte Konfigurationsparameter ausgewählt haben.

Lassen Sie uns von Grund auf neu beginnen. Spark bietet 3 Optionen zur Ausführung im Cluster: standalone, mit Mesos und mit YARN. Wir haben uns für die dritte Option entschieden, da sie für uns logisch war. Wir haben bereits einen Hadoop-Cluster. Unsere Teilnehmer sind mit seiner Architektur gut vertraut. Lassen Sie uns YARN verwenden.

spark.master=yarn

Es wird jetzt interessanter. Jede dieser 3 Bereitstellungsoptionen hat 2 Bereitstellungsmodi: client und cluster. Aus Dokumentation. verschiedenen Quellen im Internet lässt sich schließen, dass der client für interaktive Arbeiten geeignet ist — beispielsweise über Jupyter Notebook, während der cluster besser für Produktionslösungen geeignet ist. In unserem Fall interessierte uns die interaktive Arbeit, daher:

spark.deploy-mode=client

Ab diesem Zeitpunkt wird Spark grundsätzlich irgendwie auf YARN funktionieren, aber das war uns nicht genug. Da unser Programm sich mit Big Data beschäftigt, fehlte es manchmal den Teilnehmern an den Ressourcen, die im Rahmen einer gleichmäßigen Zuteilung zur Verfügung standen. Und hier entdeckten wir etwas Interessantes – die dynamische Ressourcenzuteilung. Kurz gesagt: Wenn Sie eine rechenintensive Aufgabe haben und der Cluster frei ist (zum Beispiel morgens), kann Spark Ihnen mit dieser Option zusätzliche Ressourcen bereitstellen. Der Bedarf wird durch eine ausgeklügelte Formel ermittelt. Wir werden nicht ins Detail gehen – sie funktioniert ziemlich gut.

spark.dynamicAllocation.enabled=true

Wir haben diesen Parameter gesetzt, und beim Start von Spark gab es einen Fehler, und es ließ sich nicht starten. Zurecht, denn wir hätten genauer lesen sollen. Dokumentation Dort steht, dass für eine einwandfreie Funktion noch ein zusätzlicher Parameter aktiviert werden muss.

spark.shuffle.service.enabled=true

Wozu dient es? Wenn unser Job nicht mehr so viele Ressourcen benötigt, sollte Spark diese in den allgemeinen Pool zurückgeben. Die ressourcenintensivste Phase in fast jeder MapReduce-Aufgabe ist die Shuffle-Phase. Dieser Parameter ermöglicht es, die während dieser Phase erzeugten Daten zu speichern und die Executors entsprechend freizugeben. Ein Executor ist der Prozess, der auf dem Worker alles berechnet. Er hat eine bestimmte Anzahl von Prozessorkernen und eine bestimmte Menge an Speicher.

Wir haben diesen Parameter hinzugefügt. Es scheint alles zu funktionieren. Es wurde deutlich, dass den Teilnehmern tatsächlich mehr Ressourcen zur Verfügung gestellt wurden, wenn sie benötigt wurden. Aber ein anderes Problem trat auf: Irgendwann wachten andere Teilnehmer auf und wollten ebenfalls Spark nutzen, aber alles war besetzt, und sie waren unzufrieden. Das ist verständlich. Wir begannen, die Dokumentation zu überprüfen. Dort stellte sich heraus, dass es noch einige Parameter gibt, mit denen der Prozess beeinflusst werden kann. Zum Beispiel, wenn der Executor im Wartemodus ist — nach welcher Zeit können die Ressourcen ihm entzogen werden?

spark.dynamicAllocation.executorIdleTimeout=120s

In unserem Fall gilt: Wenn Ihre Executors zwei Minuten lang keine Aktivitäten zeigen, geben Sie sie bitte in den gemeinsamen Pool zurück. Aber selbst dieser Parameter reichte manchmal nicht aus. Man konnte sehen, dass eine Person schon lange nichts mehr tat, die Ressourcen jedoch nicht freigegeben wurden. Es stellte sich heraus, dass es noch einen speziellen Parameter gibt — nach welcher Zeit man Executors abnehmen kann, die zwischengespeicherte Daten enthalten. Standardmäßig war dieser Parameter auf infinity eingestellt! Wir haben das angepasst.

spark.dynamicAllocation.cachedExecutorIdleTimeout=600s

Das bedeutet, wenn Ihre Executors in 5 Minuten keine Aktivitäten zeigen, geben Sie sie in den gemeinsamen Pool zurück. In diesem Modus wurde die Geschwindigkeit der Ressourcenerteilung für eine große Anzahl von Benutzern akzeptabel. Die Unzufriedenheit nahm ab. Aber wir beschlossen, weiterzugehen und die maximale Anzahl der Executors für eine Anwendung zu begrenzen — im Grunde für einen Teilnehmer des Programms.

spark.dynamicAllocation.maxExecutors=19

Jetzt gibt es natürlich auch Unzufriedene auf der anderen Seite — "Der Cluster steht still, und ich habe nur 19 Executors", aber was soll man machen — es braucht einen richtigen Ausgleich. Alle glücklich zu machen, wird nicht möglich sein.

Und noch eine kleine Geschichte zu den Besonderheiten unseres Falles. Bei einer praktischen Übung kamen einige Teilnehmer zu spät, und aus irgendeinem Grund ließ sich Spark nicht starten. Wir schauten uns die Anzahl der verfügbaren Ressourcen an – es schien, als wären genügend vorhanden. Spark sollte starten. Glücklicherweise war die Dokumentation bis dahin schon irgendwo im Hinterkopf gespeichert, und wir erinnerten uns, dass Spark beim Start einen Port sucht, um sich zu verbinden. Wenn der erste Port im Bereich belegt ist, greift es auf den nächsten zugreifbaren Port zurück. Wenn dieser frei ist, wird er genutzt. Zudem gibt es einen Parameter, der die maximale Anzahl an Versuchen anzeigt. Standardmäßig beträgt dieser 16. Das ist weniger als die Anzahl der Personen in unserer Gruppe während der Übung. Folglich gab Spark nach 16 Versuchen auf und meldete, dass es sich nicht starten könne. Wir haben diesen Parameter angepasst.

spark.port.maxRetries=50

Anschließend erzähle ich von einigen Einstellungen, die nicht mehr stark mit unserem speziellen Fall verbunden sind.

Für einen schnelleren Start von Spark gibt es die Empfehlung, den Ordner 'jars', der im Heimatverzeichnis SPARK_HOME liegt, zu archivieren und auf HDFS zu legen. Dadurch wird die Zeit, die für das Laden dieser JAR-Dateien auf den Workern benötigt wird, reduziert.

spark.yarn.archive=hdfs:///tmp/spark-archive.zip

Für schnellere Verarbeitung empfehlen wir, als Serializer Kryo zu verwenden. Dieser ist optimierter als der Standard-Serializer.

spark.serializer=org.apache.spark.serializer.KryoSerializer

Es gibt auch ein langjähriges Problem mit Spark, dass es oft wegen Speicherproblemen abstürzt. Dies geschieht häufig, wenn die Worker alles berechnet haben und die Ergebnisse an den Driver senden. Wir haben diesen Parameter erhöht. Standardmäßig beträgt er 1 GB, wir haben ihn auf 3 GB erhöht.

spark.driver.maxResultSize=3072

Zum Abschluss, wie man Spark auf Version 2.1 im HortonWorks-Distribution — HDP 2.5.3.0 — aktualisiert. Diese HDP-Version enthält die vorinstallierte Version 2.0, aber wir haben uns einmal entschieden, dass Spark sich aktiv weiterentwickelt und jede neue Version Fehler behebt und zusätzliche Möglichkeiten bietet, auch für die Python-API, weshalb wir beschlossen haben, ein Update durchzuführen.

Wir haben die Version von der offiziellen Website für Hadoop 2.7 heruntergeladen. Entpackt und in den HDP-Ordner verschoben. Die Symlinks richtig gesetzt. Wir starten — es startet nicht. Es wird eine sehr unklare Fehlermeldung angezeigt.

java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfig

Eine Google-Suche ergab, dass Spark beschlossen hat, nicht zu warten, bis Hadoop sich weiterentwickelt, und eine neue Version von Jersey verwenden will. Sie streiten sich darüber sogar in JIRA. Die Lösung war – herunterzuladen Jersey Version 1.17.1. Dies in den jars-Ordner von SPARK_HOME zu verschieben, wieder zip zu erstellen und auf HDFS hochzuladen.

Wir konnten diesen Fehler umgehen, aber es trat ein neuer, ziemlich vager Fehler auf.

org.apache.spark.SparkException: Die Yarn-Anwendung ist bereits beendet! Sie wurde möglicherweise beendet oder konnte den Anwendungs-Master nicht starten.

Wir versuchen die Version 2.0 zu starten – alles in Ordnung. Versuch mal herauszufinden, woran es liegt. Wir haben die Logs dieser Anwendung durchgesehen und Folgendes gesehen:

/usr/hdp/${hdp.version}/hadoop/lib/hadoop-lzo-0.6.0.${hdp.version}.jar

Kurz gesagt, aus irgendeinem Grund wurde hdp.version nicht aufgelöst. Nach einer Google-Suche fanden wir eine Lösung. Man muss in Ambari die YARN-Einstellungen aufrufen und dort einen Parameter in yarn-site hinzufügen:

hdp.version=2.5.3.0-37

Diese Magie half, und Spark lief. Wir haben einige unserer Jupyter-Notebooks getestet. Alles funktioniert. Für die erste Sitzung zu Spark am Samstag (schon morgen) sind wir bereit!

UDP. Während der Sitzung stellte sich ein weiteres Problem heraus. Irgendwann hörte YARN auf, Container für Spark bereitzustellen. In YARN musste ein Parameter angepasst werden, der standardmäßig auf 0.2 stand:

yarn.scheduler.capacity.maximum-am-resource-percent=0.8

Das bedeutet, dass nur 20 % der Ressourcen an der Verteilung beteiligt waren. Nachdem die Parameter geändert wurden, wurde YARN neu gestartet. Das Problem war behoben, und die anderen Teilnehmer konnten ebenfalls den Spark-Kontext starten.

Quelle: habr.com