Nehmen wir im Folgenden an, dass zu jedem Zeitpunkt nur ein Spark-Job ausgeführt wird.

Was ich bisher bekommen habe

Folgendes verstehe ich, was in Spark passiert:

1. Wenn a SparkContexterstellt wird, startet jeder Worker-Knoten einen Executor. Ausführende sind separate Prozesse (JVM), die eine Verbindung zum Treiberprogramm herstellen. Jeder Executor hat die JAR des Treiberprogramms. Wenn Sie einen Fahrer verlassen, werden die Testamentsvollstrecker heruntergefahren. Jeder Executor kann einige Partitionen enthalten.
2. Wenn ein Job ausgeführt wird, wird ein Ausführungsplan gemäß dem Abstammungsdiagramm erstellt.
3. Der Ausführungsjob ist in Stufen unterteilt, in denen Stufen möglichst viele benachbarte Transformationen und Aktionen (im Liniendiagramm) enthalten, jedoch keine Mischvorgänge. Somit werden Stufen durch Mischen getrennt.

ich verstehe das

• Eine Aufgabe ist ein Befehl, der vom Treiber an einen Ausführenden gesendet wird, indem das Funktionsobjekt serialisiert wird.
• Der Executor deserialisiert (mit der Treiber-JAR) den Befehl (Task) und führt ihn auf einer Partition aus.

aber

Fragen)

Wie teile ich die Bühne in diese Aufgaben auf?

Speziell:

1. Werden die Aufgaben durch die Transformationen und Aktionen bestimmt oder können sich mehrere Transformationen / Aktionen in einer Aufgabe befinden?
2. Werden die Aufgaben von der Partition bestimmt (z. B. eine Aufgabe pro Stufe und Partition)?
3. Werden die Aufgaben von den Knoten bestimmt (z. B. eine Aufgabe pro Stufe pro Knoten)?

Was ich denke (nur teilweise Antwort, auch wenn richtig)

In https://0x0fff.com/spark-architecture-shuffle wird das Mischen mit dem Bild erklärt

und ich habe den Eindruck, dass die Regel ist

Jede Stufe ist in Aufgaben mit einer Anzahl von Partitionen unterteilt, ohne Rücksicht auf die Anzahl der Knoten

Für mein erstes Bild würde ich sagen, dass ich 3 Kartenaufgaben und 3 Reduzierungsaufgaben hätte.

Für das Bild von 0x0fff würde ich sagen, dass es 8 Kartenaufgaben und 3 Reduzierungsaufgaben gibt (vorausgesetzt, es gibt nur drei orangefarbene und drei dunkelgrüne Dateien).

Offene Fragen auf jeden Fall

Ist das korrekt? Aber selbst wenn das richtig ist, werden meine obigen Fragen nicht alle beantwortet, da noch offen ist, ob sich mehrere Vorgänge (z. B. mehrere Karten) innerhalb einer Aufgabe befinden oder pro Vorgang in eine Aufgabe unterteilt sind.

Was andere sagen

Was ist eine Aufgabe in Spark? Wie führt der Spark-Worker die JAR-Datei aus? und Wie teilt der Apache Spark-Scheduler Dateien in Aufgaben auf? sind ähnlich, aber ich hatte nicht das Gefühl, dass meine Frage dort klar beantwortet wurde.

Sie haben hier eine ziemlich schöne Übersicht. Um Ihre Fragen zu beantworten

• Ein separater task nicht benötigte für jede Partition von Daten für jeden ins Leben gerufen werden stage. Bedenken Sie, dass sich jede Partition wahrscheinlich an unterschiedlichen physischen Speicherorten befindet - z. B. Blöcke in HDFS oder Verzeichnisse / Volumes für ein lokales Dateisystem.

Beachten Sie, dass die Übermittlung von Stages von der gesteuert wirdDAG Scheduler . Dies bedeutet, dass Stufen, die nicht voneinander abhängig sind, zur parallelen Ausführung an den Cluster gesendet werden können. Dies maximiert die Parallelisierungsfähigkeit des Clusters. Wenn also Vorgänge in unserem Datenfluss gleichzeitig ausgeführt werden können, werden voraussichtlich mehrere Phasen gestartet.

Wir können dies in Aktion im folgenden Spielzeugbeispiel sehen, in dem wir die folgenden Arten von Operationen ausführen:

• Laden Sie zwei Datenquellen
• Führen Sie einige Kartenoperationen für beide Datenquellen separat aus
• Schließ dich ihnen an
• Führen Sie einige Karten- und Filteroperationen für das Ergebnis aus
• Speichern Sie das Ergebnis

Wie viele Stufen werden wir dann haben?

• Jeweils 1 Stufe zum parallelen Laden der beiden Datenquellen = 2 Stufen
• Eine dritte Stufe, die das darstellt join, was abhängig ist den beiden anderen Stufen
• Hinweis: Alle Folgeoperationen, die an den verknüpften Daten arbeiten, können in derselben Phase ausgeführt werden, da sie nacheinander ausgeführt werden müssen. Das Starten zusätzlicher Stufen hat keinen Vorteil, da sie erst nach Abschluss der vorherigen Operation mit der Arbeit beginnen können.

Hier ist das Spielzeugprogramm

val sfi  = sc.textFile("/data/blah/input").map{ x => val xi = x.toInt; (xi,xi*xi) }
val sp = sc.parallelize{ (0 until 1000).map{ x => (x,x * x+1) }}
val spj = sfi.join(sp)
val sm = spj.mapPartitions{ iter => iter.map{ case (k,(v1,v2)) => (k, v1+v2) }}
val sf = sm.filter{ case (k,v) => v % 10 == 0 }
sf.saveAsTextFile("/data/blah/out")

Und hier ist die DAG des Ergebnisses

Nun: wie viele Aufgaben ? Die Anzahl der Aufgaben sollte gleich sein

Summe von ( Stage* #Partitions in the stage)

Dies könnte Ihnen helfen, verschiedene Teile besser zu verstehen:

• Phase: ist eine Sammlung von Aufgaben. Gleicher Prozess, der für verschiedene Teilmengen von Daten (Partitionen) ausgeführt wird.
• Aufgabe: Stellt eine Arbeitseinheit für eine Partition eines verteilten Datasets dar. Also in jeder Stufe, Anzahl der Aufgaben = Anzahl der Partitionen, oder wie Sie sagten "eine Aufgabe pro Stufe pro Partition".
• Jeder Executer läuft auf einem Garncontainer und jeder Container befindet sich auf einem Knoten.
• Jede Stufe verwendet mehrere Ausführende, jedem Ausführenden werden mehrere vcores zugewiesen.
• Jeder vcore kann jeweils genau eine Aufgabe ausführen
• In jedem Stadium können also mehrere Aufgaben parallel ausgeführt werden. Anzahl der ausgeführten Aufgaben = Anzahl der verwendeten vcores.

Wenn ich das richtig verstehe, gibt es zwei (verwandte) Dinge, die Sie verwirren:

1) Was bestimmt den Inhalt einer Aufgabe?

2) Was bestimmt die Anzahl der auszuführenden Aufgaben?

Der Motor von Spark "klebt" zusammen einfache Operationen auf aufeinanderfolgenden Rdds zusammen, zum Beispiel:

rdd1 = sc.textFile( ... )
rdd2 = rdd1.filter( ... )
rdd3 = rdd2.map( ... )
rdd3RowCount = rdd3.count

Wenn also rdd3 (träge) berechnet wird, generiert spark eine Aufgabe pro Partition von rdd1 und jede Aufgabe führt sowohl den Filter als auch die Zuordnung pro Zeile aus, um rdd3 zu erhalten.

Die Anzahl der Aufgaben wird durch die Anzahl der Partitionen bestimmt. Jedes RDD hat eine definierte Anzahl von Partitionen. Bei einer Quell-RDD, die aus HDFS gelesen wird (z. B. mit sc.textFile (...)), entspricht die Anzahl der Partitionen der Anzahl der vom Eingabeformat generierten Teilungen. Einige Operationen auf RDDs können zu einer RDD mit einer anderen Anzahl von Partitionen führen:

rdd2 = rdd1.repartition( 1000 ) will result in rdd2 having 1000 partitions ( regardless of how many partitions rdd1 had ).

Ein weiteres Beispiel sind Joins:

rdd3 = rdd1.join( rdd2  , numPartitions = 1000 ) will result in rdd3 having 1000 partitions ( regardless of partitions number of rdd1 and rdd2 ).

(Die meisten) Operationen, die die Anzahl der Partitionen ändern, beinhalten ein Mischen. Wenn wir zum Beispiel Folgendes tun:

rdd2 = rdd1.repartition( 1000 ) 

Was tatsächlich passiert, ist, dass die Aufgabe auf jeder Partition von rdd1 eine Endausgabe erzeugen muss, die in der folgenden Phase gelesen werden kann, damit rdd2 genau 1000 Partitionen hat (wie sie es tun? Hash oder Sortieren ). Aufgaben auf dieser Seite werden manchmal als "Kartenaufgaben (Seitenaufgaben)" bezeichnet. Eine Aufgabe, die später auf rdd2 ausgeführt wird, wirkt sich auf eine Partition (von rdd2!) Aus und muss herausfinden, wie die für diese Partition relevanten kartenseitigen Ausgaben gelesen / kombiniert werden. Aufgaben auf dieser Seite werden manchmal als "(Neben-) Aufgaben reduzieren" bezeichnet.

Die beiden Fragen hängen zusammen: Die Anzahl der Aufgaben in einer Stufe ist die Anzahl der Partitionen (die den aufeinanderfolgenden "zusammengeklebten" Festplatten gemeinsam sind), und die Anzahl der Partitionen einer Festplatte kann sich zwischen den Stufen ändern (indem die Anzahl der Partitionen für einige angegeben wird mischen, was zum Beispiel den Betrieb verursacht).

Sobald die Ausführung einer Stufe beginnt, können ihre Aufgaben Aufgabenplätze belegen. Die Anzahl der gleichzeitigen Task-Slots ist numExecutors * ExecutorCores. Im Allgemeinen können diese mit Aufgaben aus verschiedenen, nicht abhängigen Phasen belegt werden.