Wie verwaltet Spark Stadien?

Question

Ich versuche zu verstehen, wie Jobs und Stufen in Spark definiert sind, und dafür verwende ich jetzt den Code, den ich gefunden here und UI funken. Um es auf Spark UI zu sehen, musste ich den Text mehrmals kopieren und einfügen, so dass es mehr Zeit für die Verarbeitung benötigt. Hier

ist der Ausgang der Funken UI:

Nun, ich verstehe, dass es drei Arbeitsplätze, weil es drei Aktionen sind und auch, dass die Stufen erzeugt werden, weil die Shuffle-Aktionen, aber was ich Verstehst du nicht, warum in Job 1 die Stufen 4, 5 und 6 die gleichen sind wie die Stufen 0, 1 und 2 von Job 0 und das gleiche auch für Job 2.

Wie kann ich wissen, in welchen Etappen es sein wird? mehr als ein Job, der nur den Java-Code sieht (bevor irgendetwas ausgeführt wird)? Und warum werden Stufe 4 und 9 übersprungen und wie kann ich wissen, dass es vor der Ausführung passieren wird?

Answer 1

Ich verstehe, dass es drei Arbeitsplätze, weil es drei Aktionen

ich auch sagen würde, dass es mehr Spark-Arbeitsplätze haben könnte, aber die Mindestanzahl ist 3. Alles hängt von der Implementierung der Transformationen und der verwendeten Aktion ab.

ich nicht verstehe, ist, warum in den Job 1 Stufen 4, 5 und 6 sind die gleichen wie Stufen 0, 1 und 2 von Job 0 und das gleiche geschieht für Job 2.

Job 1 ist das Ergebnis einer Aktion, die auf einer RDD ausgeführt wurde, finalRdd. Diese RDD wurde unter Verwendung (rückwärts) erstellt: join, textFile, map und distinct.

val people = sc.textFile("people.csv").map { line => 
    val tokens = line.split(",") 
    val key = tokens(2) 
    (key, (tokens(0), tokens(1))) }.distinct 

val cities = sc.textFile("cities.csv").map { line => 
    val tokens = line.split(",") 
    (tokens(0), tokens(1)) 
} 

val finalRdd = people.join(cities) 

Führen Sie die oben genannten und Sie werden die gleiche DAG sehen.

Wenn Sie nun leftOuterJoin oder rightOuterJoin Aktionen ausführen, werden Sie die beiden anderen DAGs bekommen. Sie verwenden die zuvor verwendeten RDDs, um neue Spark-Jobs auszuführen, und Sie sehen dieselben Phasen.

warum Stufe 4 und 9 übersprungen

Oft Spark Ausführung einiger Stufen überspringen. Die ausgegrauten Stufen sind bereits berechnet, damit Spark sie wiederverwenden kann und so die Leistung verbessert.

Wie kann ich wissen, welche Phasen in mehr als ein Job sein wird, nur den Java-Code zu sehen (vor irgendetwas ausgeführt wird)?

Das ist, was RDD-Linie (Graph) bietet.

scala> people.leftOuterJoin(cities).toDebugString 
res15: String = 
(3) MapPartitionsRDD[99] at leftOuterJoin at <console>:28 [] 
| MapPartitionsRDD[98] at leftOuterJoin at <console>:28 [] 
| CoGroupedRDD[97] at leftOuterJoin at <console>:28 [] 
+-(2) MapPartitionsRDD[81] at distinct at <console>:27 [] 
| | ShuffledRDD[80] at distinct at <console>:27 [] 
| +-(2) MapPartitionsRDD[79] at distinct at <console>:27 [] 
|  | MapPartitionsRDD[78] at map at <console>:24 [] 
|  | people.csv MapPartitionsRDD[77] at textFile at <console>:24 [] 
|  | people.csv HadoopRDD[76] at textFile at <console>:24 [] 
+-(3) MapPartitionsRDD[84] at map at <console>:29 [] 
    | cities.csv MapPartitionsRDD[83] at textFile at <console>:29 [] 
    | cities.csv HadoopRDD[82] at textFile at <console>:29 [] 

Wie Sie selbst sehen können, werden Sie sich mit 4 Stufen enden, da es 3 Shuffle Abhängigkeiten (die Ränder mit den Zahlen von Partitionen).

Zahlen in den runden Klammern sind die Anzahl der Partitionen, die DAGScheduler schließlich zum Erstellen von Aufgabensätzen mit der genauen Anzahl von Aufgaben verwenden. Ein TaskSet pro Stufe.