Apache Spark: Die Anzahl der Kerne im Vergleich zur Anzahl der Ausführenden

192

Ich versuche, die Beziehung zwischen der Anzahl der Kerne und der Anzahl der Ausführenden zu verstehen, wenn ein Spark-Job auf YARN ausgeführt wird.

Die Testumgebung ist wie folgt:

  • Anzahl der Datenknoten: 3
  • Datenknoten-Maschinenspezifikation:
    • CPU: Core i7-4790 (Anzahl der Kerne: 4, Anzahl der Threads: 8)
    • RAM: 32 GB (8 GB x 4)
    • Festplatte: 8 TB (2 TB x 4)

  • Netzwerk: 1 GB

  • Spark-Version: 1.0.0

  • Hadoop-Version: 2.4.0 (Hortonworks HDP 2.1)

  • Spark-Jobfluss: sc.textFile -> Filter -> Map -> Filter -> MapToPair -> ReduceByKey -> Map -> SaveAsTextFile

  • Eingabedaten

    • Typ: einzelne Textdatei
    • Größe: 165 GB
    • Anzahl der Zeilen: 454.568.833

  • Ausgabe

    • Anzahl der Zeilen nach dem zweiten Filter: 310.640.717
    • Anzahl der Zeilen der Ergebnisdatei: 99.848.268
    • Größe der Ergebnisdatei: 41 GB

Der Job wurde mit folgenden Konfigurationen ausgeführt:

  1. --master yarn-client --executor-memory 19G --executor-cores 7 --num-executors 3 (Ausführende pro Datenknoten verwenden so viel wie Kerne)

  2. --master yarn-client --executor-memory 19G --executor-cores 4 --num-executors 3 (Anzahl der Kerne reduziert)

  3. --master yarn-client --executor-memory 4G --executor-cores 2 --num-executors 12 (weniger Kern, mehr Vollstrecker)

Verstrichene Zeiten:

  1. 50 min 15 sek

  2. 55 min 48 sek

  3. 31 min 23 sek

Zu meiner Überraschung war (3) viel schneller.
Ich dachte, dass (1) schneller sein würde, da es beim Mischen weniger Kommunikation zwischen Ausführenden geben würde.
Obwohl die Anzahl der Kerne von (1) kleiner als (3) ist, ist die Anzahl der Kerne nicht der Schlüsselfaktor, da 2) eine gute Leistung erbracht hat.

(Nach der Antwort von pwilmot wurden folgende hinzugefügt.)

Zur Information lautet die Bildschirmaufnahme des Leistungsmonitors wie folgt:

  • Ganglia-Datenknotenübersicht für (1) - Job gestartet um 04:37.

Ganglia-Datenknotenübersicht für (1)

  • Ganglia-Datenknotenübersicht für (3) - Job gestartet um 19:47. Bitte ignorieren Sie die Grafik vor diesem Zeitpunkt.

Zusammenfassung der Gangliendatenknoten für (3)

Das Diagramm ist grob in zwei Abschnitte unterteilt:

  • Erstens: von Anfang bis zur ReduzierungByKey: CPU-intensiv, keine Netzwerkaktivität
  • Zweitens: Nach dem Reduzieren von ByKey: CPU wird die Netzwerk-E / A abgeschlossen.

Wie die Grafik zeigt, kann (1) so viel CPU-Leistung verbrauchen, wie angegeben wurde. Es könnte also nicht das Problem der Anzahl der Threads sein.

Wie kann man dieses Ergebnis erklären?

hadoop apache-spark yarn zeodtr
quelle
2
Jetzt vermute ich GC ... Tatsächlich ist auf der Spark-Benutzeroberfläche die Gesamtzeit für GC länger als 1) als 2).
Zeodtr
Warum hast du nicht 3) mit 19G versucht? Könnte es sein, dass die Beschränkung der Arbeiter auf 4G den NUMA-Effekt verringert, den einige Leute haben? Das heißt, Ihr 4G befindet sich auf einem der 2 Kerne, die Ihrem Workflow zugeordnet sind, und daher kommt es zu einer geringeren E / A-Verlangsamung, was zu einer besseren Gesamtleistung führt. Ansonsten denke ich, dass eine Hauptfrage lautet: Wie viele Kerne / Threads können einen einzelnen Executor für einen Worker verwenden? (Man kann nur die Gesamtzahl der Kerne für einen Arbeiter angeben, nicht die Granularität des Testamentsvollstreckers)
Bacon
4
Übrigens habe ich gerade den Code in core / src / main / scala / org / apache / spark / deploy / worker / ExecutorRunner.scala überprüft und es scheint, dass 1 Executor = 1 Worker-Thread ist.
Bacon
etwas spät, aber hier ist ein Beitrag über Cloudera zu diesem Thema: blog.cloudera.com/blog/2015/03/…
Orelus
1
Übrigens fand ich diese Informationen in einem Cloudera Slide Deck Slideshare.net/cloudera/… , das ein wenig über die Entscheidungsfindung in Executoren, Kernen und Speicher erklärt
Manish Sahni

Antworten:

58

Um dies hoffentlich etwas konkreter zu gestalten, finden Sie hier ein Beispiel für die Konfiguration einer Spark-App, um so viel wie möglich vom Cluster zu verwenden: Stellen Sie sich einen Cluster mit sechs Knoten vor, auf denen NodeManager ausgeführt werden, die jeweils mit 16 Kernen und 64 GB Speicher ausgestattet sind . Die NodeManager-Kapazitäten yarn.nodemanager.resource.memory-mb und yarn.nodemanager.resource.cpu-vcores sollten wahrscheinlich auf 63 * 1024 = 64512 (Megabyte) bzw. 15 eingestellt sein. Wir vermeiden es, YARN-Containern 100% der Ressourcen zuzuweisen, da der Knoten einige Ressourcen benötigt, um die OS- und Hadoop-Daemons auszuführen. In diesem Fall belassen wir ein Gigabyte und einen Kern für diese Systemprozesse. Cloudera Manager hilft, indem es diese berücksichtigt und diese YARN-Eigenschaften automatisch konfiguriert.

Der wahrscheinlich erste Impuls wäre, --num-executors 6 --executor-cores 15 --executor-memory 63G zu verwenden . Dies ist jedoch der falsche Ansatz, weil:

63 GB + Der Overhead des Executor-Speichers passt nicht in die 63 GB-Kapazität der NodeManager. Der Anwendungsmaster nimmt einen Kern auf einem der Knoten ein, was bedeutet, dass auf diesem Knoten kein Platz für einen 15-Kern-Executor vorhanden ist. 15 Kerne pro Executor können zu einem schlechten HDFS-E / A-Durchsatz führen.

Eine bessere Option wäre die Verwendung von --num-executors 17 --executor-cores 5 --executor-memory 19G . Warum?

Diese Konfiguration führt zu drei Executoren auf allen Knoten mit Ausnahme des Knotens mit dem AM, der zwei Executoren hat. --executor-memory wurde abgeleitet als (63/3 Executor pro Knoten) = 21. 21 * 0.07 = 1.47. 21 - 1,47 ~ 19.

Die Erklärung wurde in einem Artikel in Clouderas Blog " How-to: Tune Your Apache Spark Jobs" (Teil 2) gegeben .

DzOrdre
quelle
1
"Diese Konfiguration führt zu drei Executoren auf allen Knoten mit Ausnahme des Knotens mit dem AM, der zwei Executoren haben wird." Was bedeutet das für "--executor-cores 5"?
Derek
Dies bedeutet, dass jeder Executor 5 Kerne verwendet. Jeder Knoten hat 3 Executoren, die 15 Kerne verwenden, außer dass auf einem der Knoten auch der Anwendungsmaster für den Job ausgeführt wird, sodass nur 2 Executoren gehostet werden können, dh 10 Kerne, die als Executoren verwendet werden.
Davos
Schön erklärt - bitte beachten Sie, dass dies für yarn.scheduler.capacity.resource-calculatorBehinderte gilt, was die Standardeinstellung ist. Dies liegt daran, dass die Zeitplanung standardmäßig nach Speicher und nicht nach CPU erfolgt.
YoYo
1
Mehr Executoren können zu einem schlechten HDFS-E / A-Durchsatz führen. Wenn ich also überhaupt kein HDFS verwende, kann ich in diesem Fall mehr als 5 Kerne pro Executor verwenden?
Darshan
Ich denke, der Anwendungsmaster läuft auf jedem Knoten. Wie oben beschrieben, bedeutet dies, dass nur 1 Anwendungsmaster zum Ausführen des Jobs vorhanden ist. Ist das korrekt?
Roshan Fernando
15

Laut Sandy Ryza läuft Ihre Spark-App auf HDFS

Ich habe festgestellt, dass der HDFS-Client Probleme mit Tonnen von gleichzeitigen Threads hat. Eine grobe Vermutung ist, dass höchstens fünf Aufgaben pro Executor den vollen Schreibdurchsatz erzielen können. Daher ist es gut, die Anzahl der Kerne pro Executor unter dieser Anzahl zu halten.

Daher glaube ich, dass Ihre erste Konfiguration aufgrund des schlechten HDFS-E / A-Durchsatzes langsamer ist als die dritte

tgbaggio
quelle
11

Ich habe selbst nicht mit diesen Einstellungen gespielt, daher handelt es sich nur um Spekulationen. Wenn wir dieses Problem jedoch als normale Kerne und Threads in einem verteilten System betrachten, können Sie in Ihrem Cluster bis zu 12 Kerne (4 * 3 Computer) und 24 Threads verwenden (8 * 3 Maschinen). In Ihren ersten beiden Beispielen geben Sie Ihrem Job eine angemessene Anzahl von Kernen (potenzieller Rechenraum), aber die Anzahl der Threads (Jobs), die auf diesen Kernen ausgeführt werden sollen, ist so begrenzt, dass Sie nicht viel von der zugewiesenen Verarbeitungsleistung verwenden können und daher ist der Job langsamer, obwohl mehr Rechenressourcen zugewiesen sind.

Sie erwähnen, dass Ihr Anliegen im Shuffle-Schritt lag. Obwohl es hilfreich ist, den Overhead im Shuffle-Schritt zu begrenzen, ist es im Allgemeinen viel wichtiger, die Parallelisierung des Clusters zu nutzen. Denken Sie an den Extremfall - ein Single-Threaded-Programm ohne Shuffle.

pwilmot
quelle
Vielen Dank für Ihre Antwort. Aber ich vermute, dass die Anzahl der Threads nicht das Hauptproblem ist. Ich habe die Überwachungsbildschirmaufnahme hinzugefügt. Wie die Grafik zeigt, kann 1) so viel CPU-Leistung verbrauchen, wie angegeben wurde.
Zeodtr
1
@zeodtr pwilmot ist korrekt - Sie benötigen mindestens 2-4 Aufgaben, um das volle Potenzial Ihrer Kerne auszuschöpfen. Sagen wir das war - ich verwende normalerweise mindestens 1000 Partitionen für meinen 80-Kern-Cluster.
Samthebest
@samthebest Was ich wissen möchte, ist der Grund für den Leistungsunterschied zwischen 1) und 3). Wenn ich mir die Spark-Benutzeroberfläche ansehe, werden in Abschnitt 2 beide Aufgaben parallel ausgeführt (warum 21 statt 24 bei 3), ist derzeit nicht bekannt.) Die Aufgaben für 3) werden jedoch nur schneller ausgeführt.
Zeodtr
10

Kurze Antwort : Ich denke, tgbaggio ist richtig. Sie haben die HDFS-Durchsatzgrenzen Ihrer Executoren erreicht.

Ich denke, die Antwort hier ist möglicherweise etwas einfacher als einige der Empfehlungen hier.

Der Hinweis für mich ist im Cluster-Netzwerkdiagramm. Für Lauf 1 liegt die Auslastung konstant bei ~ 50 MByte / s. Für Lauf 3 wird die stetige Auslastung verdoppelt, etwa 100 MByte / s.

Von der cloudera Blog - Post von gemeinsamen DzOrd , können Sie dieses wichtige Angebot finden Sie unter :

Ich habe festgestellt, dass der HDFS-Client Probleme mit Tonnen von gleichzeitigen Threads hat. Eine grobe Vermutung ist, dass höchstens fünf Aufgaben pro Executor den vollen Schreibdurchsatz erzielen können. Daher ist es gut, die Anzahl der Kerne pro Executor unter dieser Anzahl zu halten.

Lassen Sie uns ein paar Berechnungen durchführen, um zu sehen, welche Leistung wir erwarten, wenn dies zutrifft.

Führen Sie 1: 19 GB, 7 Kerne und 3 Executoren aus

  • 3 Executoren x 7 Threads = 21 Threads
  • Bei 7 Kernen pro Executor erwarten wir eine begrenzte Anzahl von E / A auf HDFS (maximal ~ 5 Kerne).
  • effektiver Durchsatz ~ = 3 Executoren x 5 Threads = 15 Threads

Führen Sie 3: 4 GB, 2 Kerne und 12 Executoren aus

  • 2 Executoren x 12 Threads = 24 Threads
  • 2 Kerne pro Executor, daher ist der HDFS-Durchsatz in Ordnung
  • effektiver Durchsatz ~ = 12 Executoren x 2 Threads = 24 Threads

Wenn der Job zu 100% durch Parallelität begrenzt ist (Anzahl der Threads). Wir würden erwarten, dass die Laufzeit perfekt umgekehrt mit der Anzahl der Threads korreliert.

ratio_num_threads = nthread_job1 / nthread_job3 = 15/24 = 0.625
inv_ratio_runtime = 1/(duration_job1 / duration_job3) = 1/(50/31) = 31/50 = 0.62

So ratio_num_threads ~= inv_ratio_runtime , und es sieht , als wären wir netzwerkbeschränkt.

Der gleiche Effekt erklärt den Unterschied zwischen Lauf 1 und Lauf 2.

Führen Sie 2: 19 GB, 4 Kerne und 3 Executoren aus

  • 3 Executoren x 4 Threads = 12 Threads
  • mit 4 Kernen pro Executor, ok IO zu HDFS
  • effektiver Durchsatz ~ = 3 Executoren x 4 Threads = 12 Threads

Vergleichen der Anzahl der effektiven Threads und der Laufzeit:

ratio_num_threads = nthread_job2 / nthread_job1 = 12/15 = 0.8
inv_ratio_runtime = 1/(duration_job2 / duration_job1) = 1/(55/50) = 50/55 = 0.91

Es ist nicht so perfekt wie der letzte Vergleich, aber wir sehen immer noch einen ähnlichen Leistungsabfall, wenn wir Threads verlieren.

Nun zum letzten Punkt: Warum ist es so, dass wir mit mehr Threads eine bessere Leistung erzielen, insb. mehr Threads als die Anzahl der CPUs?

Eine gute Erklärung für den Unterschied zwischen Parallelität (was wir durch Aufteilen von Daten auf mehrere CPUs erhalten) und Parallelität (was wir erhalten, wenn wir mehrere Threads verwenden, um an einer einzelnen CPU zu arbeiten) finden Sie in diesem großartigen Beitrag von Rob Pike: Concurrency ist keine Parallelität .

Die kurze Erklärung lautet: Wenn ein Spark-Job mit einem Dateisystem oder Netzwerk interagiert, verbringt die CPU viel Zeit damit, auf die Kommunikation mit diesen Schnittstellen zu warten und nicht viel Zeit damit zu verbringen, tatsächlich "zu arbeiten". Indem Sie diesen CPUs mehr als eine Aufgabe gleichzeitig zur Verfügung stellen, verbringen sie weniger Zeit mit Warten und mehr Zeit mit Arbeiten, und Sie sehen eine bessere Leistung.

turtlemonvh
quelle
1
Interessante und überzeugende Erklärung, ich frage mich, wie Sie zu Ihrer Vermutung gekommen sind, dass der Executor maximal 5 Aufgaben hat, um einen maximalen Durchsatz zu erzielen.
Dat Nguyen
Die Nummer 5 ist also nichts, was ich mir ausgedacht habe: Ich habe gerade Anzeichen von E / A-Engpässen bemerkt und bin losgefahren, um herauszufinden, woher diese Engpässe kommen könnten.
Turtlemonvh
8

Aus den hervorragenden Ressourcen, die auf der Sparklyr-Paketseite von RStudio verfügbar sind :

SPARK DEFINITIONEN :

Es kann nützlich sein, einige einfache Definitionen für die Spark-Nomenklatur anzugeben:

Knoten : Ein Server

Arbeiterknoten : Ein Server, der Teil des Clusters ist und zum Ausführen von Spark-Jobs verfügbar ist

Hauptknoten : Der Server, der die Arbeiter Knoten koordiniert.

Testamentsvollstrecker : Eine Art virtuelle Maschine innerhalb eines Knotens. Ein Knoten kann mehrere Executoren haben.

Treiberknoten : Der Knoten, der die Spark-Sitzung initiiert. In der Regel ist dies der Server, auf dem sich sparklyr befindet.

Treiber (Executor) : Der Treiberknoten wird auch in der Executor-Liste angezeigt.

d8aninja
quelle
1

Die dynamische Zuweisung von Spark bietet Flexibilität und weist Ressourcen dynamisch zu. In dieser Anzahl von Min- und Max-Executoren können angegeben werden. Es kann auch die Anzahl der Executoren angegeben werden, die zu Beginn der Anwendung gestartet werden müssen.

Lesen Sie unten auf dem gleichen:

http://spark.apache.org/docs/latest/configuration.html#dynamic-allocation

Harikrishnan Ck
quelle
1

Es gibt ein kleines Problem in den ersten beiden Konfigurationen, denke ich. Die Konzepte von Threads und Kernen wie folgt. Das Konzept des Threading ist, wenn die Kerne ideal sind, dann verwenden Sie diesen Kern, um die Daten zu verarbeiten. Daher ist der Speicher in den ersten beiden Fällen nicht voll ausgelastet. Wenn Sie dieses Beispiel als Benchmark verwenden möchten, wählen Sie die Maschinen mit mehr als 10 Kernen pro Maschine aus. Dann machen Sie die Benchmark.

Geben Sie jedoch nicht mehr als 5 Kerne pro Executor an, da die I / O-Leistung einen Flaschenhals aufweist.

Die besten Maschinen für dieses Benchmarking sind möglicherweise Datenknoten mit 10 Kernen.

Datenknoten-Maschinenspezifikation: CPU: Core i7-4790 (Anzahl der Kerne: 10, Anzahl der Threads: 20) RAM: 32 GB (8 GB x 4) Festplatte: 8 TB (2 TB x 4)

loneStar
quelle
0

Ich denke, einer der Hauptgründe ist die Lokalität. Ihre Eingabedateigröße beträgt 165 GB, die zugehörigen Blöcke der Datei sind sicherlich auf mehrere DataNodes verteilt. Mehr Ausführende können eine Netzwerkkopie vermeiden.

Versuchen Sie, die Anzahl der Blöcke für Executor gleich zu setzen. Ich denke, das kann schneller sein.

zwb
quelle