Konfiguration von Spark auf YARN

Khabr, hallo! Gestern auf dem Meetup, das sich mit Apache Spark beschäftigte, gab es viele Fragen von den Teilnehmern, die mit der Konfiguration dieses Tools zu tun hatten. Wir haben beschlossen, unser Wissen darüber zu teilen. Das Thema ist nicht einfach, deshalb schlagen wir vor, auch in den Kommentaren Erfahrungen auszutauschen. Vielleicht verstehen wir auch nicht alles richtig oder verwenden es falsch.

Eine kurze Einführung – wie wir Spark verwenden. Wir haben ein dreimonatiges Programm „Spezialist für große Daten“, und im gesamten zweiten Modul arbeiten unsere Teilnehmer mit diesem Tool. Unsere Aufgabe als Veranstalter ist es, einen Cluster für die Verwendung in einem solchen Fall vorzubereiten.

Das Besondere an unserer Nutzung ist, dass die Anzahl der Personen, die gleichzeitig mit Spark arbeiten, gleich der gesamten Gruppe sein kann. Beispielsweise beim Seminar, wenn alle gleichzeitig etwas ausprobieren und das Nachmachen, was unser Dozent zeigt. Das sind manchmal fast 40 Personen. Wahrscheinlich gibt es nicht viele Unternehmen auf der Welt, die mit diesem Nutzungsszenario konfrontiert sind.

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

Lass uns ganz von vorne beginnen. Spark hat 3 Möglichkeiten, um auf einem Cluster zu arbeiten: standalone, mit Mesos und mit YARN. Wir haben uns für die dritte Option entschieden, weil 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

Dann wird es interessanter. Jede dieser 3 Bereitstellungsoptionen hat 2 Bereitstellungsarten: client und cluster. Ausgehend von Dokumentation und verschiedenen Links im Internet kann man schließen, dass client für interaktive Arbeiten geeignet ist — beispielsweise über Jupyter Notebook, während cluster mehr für Produktionslösungen geeignet ist. In unserem Fall interessierte uns die interaktive Arbeit, deshalb:

spark.deploy-mode=client

Im Allgemeinen wird Spark ab diesem Zeitpunkt bereits auf YARN funktionieren, aber das war uns nicht genug. Da wir ein Programm über große Daten haben, fehlte es den Teilnehmern manchmal an den Ressourcen, die bei einer gleichmäßigen Verteilung zur Verfügung standen. Und hier fanden wir eine interessante Sache – die dynamische Ressourcenallokation. Kurz gesagt, der Kern ist folgender: Wenn Sie eine rechenintensive Aufgabe haben und der Cluster frei ist (zum Beispiel morgens), dann kann Spark Ihnen mit dieser Option zusätzliche Ressourcen zuteilen. Der Bedarf wird dort durch eine komplizierte Formel berechnet. Wir wollen ins Detail nicht eingehen – sie funktioniert ganz gut.

spark.dynamicAllocation.enabled=true

Wir haben diesen Parameter gesetzt, und beim Starten von Spark gibt es einen Fehler und es startet nicht. Richtig, denn wir hätten lesen sollen Dokumentation aufmerksamer. Dort steht, dass für alles in Ordnung ist, noch ein zusätzlicher Parameter aktiviert werden muss.

spark.shuffle.service.enabled=true

Warum ist das nötig? Wenn unser Job nicht mehr so viele Ressourcen benötigt, sollte Spark sie in den gemeinsamen Pool zurückgeben. Der zeitaufwändigste Schritt in fast jeder MapReduce-Aufgabe ist die Shuffle-Phase. Dieser Parameter ermöglicht es, die Daten, die in dieser Phase entstehen, zu speichern und damit die Executors freizugeben. Ein Executor ist der Prozess, der auf dem Worker alles berechnet. Er hat eine bestimmte Anzahl von CPU-Kernen und einen bestimmten Speicher.

Wir haben diesen Parameter hinzugefügt. Es schien 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 kann man nachvollziehen. Wir begannen, in der Dokumentation nachzusehen. Dort stellte sich heraus, dass es noch eine Reihe von Parametern gibt, mit denen man den Prozess beeinflussen kann. Zum Beispiel: Wenn der Executor im Leerlauf ist – nach wie viel Zeit können wir ihm die Ressourcen entziehen?

spark.dynamicAllocation.executorIdleTimeout=120s

In unserem Fall — wenn Ihre Executors in zwei Minuten nichts tun, dann, seien Sie so freundlich, sie zurück in den allgemeinen Pool zu geben. Aber auch dieser Parameter war nicht immer ausreichend. Es war deutlich zu sehen, dass die Person schon lange nichts macht, und die Ressourcen wurden nicht freigegeben. Es stellte sich heraus, dass es noch einen speziellen Parameter gibt — nach welcher Zeit die Executors ausgewählt werden sollen, die zwischengespeicherte Daten enthalten. Standardmäßig war dieser Parameter — unendlich! Wir haben ihn angepasst.

spark.dynamicAllocation.cachedExecutorIdleTimeout=600s

Das heißt, wenn Ihre Executors in den letzten 5 Minuten nichts tun, geben Sie sie zurück in den allgemeinen Pool. In diesem Modus wurde die Geschwindigkeit der Freigabe und Bereitstellung von Ressourcen für eine große Anzahl von Benutzern akzeptabel. Die Anzahl der Unzufriedenen hat abgenommen. Aber wir haben uns entschieden, 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

Nun gab es natürlich Beschwerden von anderer Seite — „Der Cluster steht still, und ich habe nur 19 Executors”, aber was soll's — es braucht irgendein richtiges Gleichgewicht. Es ist nicht möglich, alle glücklich zu machen.

Und noch eine kleine Geschichte, die mit den Besonderheiten unseres Falls zusammenhängt. Einmal kamen einige Personen zu spät zum praktischen Unterricht, und aus irgendeinem Grund startete Spark nicht. Wir haben die Anzahl der verfügbaren Ressourcen angeschaut — schien ausreichend zu sein. Spark sollte starten. Glücklicherweise hatte sich zu diesem Zeitpunkt die Dokumentation bereits irgendwo ins Gedächtnis eingeprägt, und wir erinnerten uns, dass Spark beim Start einen Port sucht, auf dem es läuft. Wenn der erste Port im Bereich belegt ist, wechselt es zum nächsten. Wenn dieser frei ist, nimmt es ihn. Und es gibt einen Parameter, der die maximale Anzahl der Versuche dafür angibt. Standardmäßig sind es 16. Das ist weniger als die Anzahl der Personen in unserer Gruppe beim Unterricht. Folglich brach Spark nach 16 Versuchen ab und sagte, dass es nicht starten könnte. Wir haben diesen Parameter angepasst.

spark.port.maxRetries=50

Ich werde nun über einige Einstellungen sprechen, die nicht mehr stark mit den Besonderheiten unseres Falls verbunden sind.

Um Spark schneller zu starten, gibt es die Empfehlung, den Jar-Ordner, der im Stammverzeichnis von SPARK_HOME liegt, zu archivieren und auf HDFS zu legen. Dadurch wird die Zeit für das Laden dieser JARs auf den Workern eingespart.

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

Um die Leistung zu verbessern, wird empfohlen, als Serialisierer Kryo zu verwenden. Es ist optimierter als der Standardserialisierer.

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

Ein weiteres bekanntes Problem mit Spark ist, dass es häufig aufgrund von Speichermangel abstürzt. Dies geschieht oft, wenn die Worker alle Berechnungen abgeschlossen haben und das Ergebnis 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

Und schließlich, als Bonus. Wie man Spark auf Version 2.1 im HortonWorks-Distro – HDP 2.5.3.0 – aktualisiert. Diese HDP-Version enthält eine vorinstallierte Version 2.0, aber wir haben irgendwann entschieden, dass sich Spark ziemlich aktiv weiterentwickelt, und jede neue Version behebt Bugs und bietet zusätzliche Funktionen, auch für die Python-API. Daher haben wir uns entschlossen, das Update durchzuführen.

Wir haben die Version von der offiziellen Website für Hadoop 2.7 heruntergeladen. Wir haben sie entpackt und in den HDP-Ordner verschoben. Wir haben die Symlinks wie gewünscht erstellt. Wir starten – es funktioniert nicht. Es gibt eine sehr unklare Fehlermeldung.

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

Nach einer Google-Suche stellten wir fest, dass Spark beschlossen hat, nicht zu warten, bis Hadoop eine neue Version bereitstellt, und eine neue Version von Jersey verwenden wollte. Sie streiten sich darüber in JIRA. Die Lösung war – herunterzuladen Jersey Version 1.17.1. Dieses in den jars-Ordner von SPARK_HOME verschieben, wieder zippen und auf HDFS hochladen.

Wir haben diesen Fehler umgangen, aber ein neuer und recht vager trat auf.

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

Wir versuchen dabei, Version 2.0 zu starten - alles in Ordnung. Versuche zu erraten, woran es liegt. Wir haben die Protokolle dieser Anwendung eingesehen und Folgendes gesehen:

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

Im Allgemeinen wurde aus irgendeinem Grund hdp.version nicht aufgelöst. Nach einer Google-Suche fanden wir die Lösung. Man muss in Ambari zu den YARN-Einstellungen gehen und dort einen Parameter im benutzerdefinierten yarn-site hinzufügen:

hdp.version=2.5.3.0-37

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

UPD. Während des Unterrichts stellte sich ein weiteres Problem heraus. Irgendwann hörte YARN auf, Container für Spark auszugeben. In YARN musste der Parameter geändert werden, der standardmäßig auf 0,2 eingestellt war:

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

Das heißt, nur 20 % der Ressourcen waren an der Ressourcenverteilung beteiligt. Nach Anpassung der Parameter haben wir YARN neu geladen. Das Problem wurde gelöst, und auch die anderen Teilnehmer konnten den Spark-Kontext starten.

Quelle: habr.com