Как справиться с этапами?

Question

Я пытаюсь понять, как задания и этапы определены в искры, и для этого я теперь использую код, который я нашел here и искровой интерфейс. Чтобы увидеть это на искровом UI, мне пришлось копировать и вставлять текст в файлы несколько раз, поэтому для обработки требуется больше времени.

Вот выход искрового UI:

Теперь я понимаю, что есть три рабочих мест, потому что есть три действия, а также, что стадия образуется из-за перетасовки действий, но то, что я не понимаю, почему на этапах 1-го задания 4, 5 и 6 такие же, как и этапы 0, 1 и 2 задания 0, и то же самое происходит и для задания 2.

Как я могу узнать, какие этапы будут в больше, чем работа, только видя код Java (перед выполнением чего-либо)? А также, почему этапы 4 и 9 пропущены, и как я могу знать, что это произойдет до выполнения?

Answer 1

Я понимаю, что есть три рабочих мест, потому что есть три действия

Я бы даже сказать, что могло бы быть больше Спарк рабочих мест, но минимальное число 3. Все зависит от реализации преобразований и используемого действия.

Я не понимаю, почему в работе 1 ступени 4, 5 и 6 являются такими же, как этапы 0, 1 и 2 Иова 0 и то же самое происходит для задания 2.

Работа 1 является результатом действия, которое выполнялось на RDD, finalRdd. Эта RDD была создана с использованием (в обратном порядке): join, textFile, map и 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) 

Завершите вышеуказанное, и вы увидите тот же самый DAG.

Теперь, когда вы выполняете leftOuterJoin или rightOuterJoin действия, вы получите две другие группы DAG. Вы используете ранее используемые RDD для запуска новых заданий Spark, и вы увидите те же самые этапы.

почему стадия 4 и 9 пропущено

Часто Спарк будет пропускать выполнение некоторых этапов. Сегментированные этапы - это те, которые уже вычислены, поэтому Spark будет их повторно использовать и таким образом повысить производительность.

Как я могу знать, какие этапы будет более, чем работа только видящего код Java (перед выполнением ничего)?

Это то, что предлагает линия RDD (график).

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 [] 

Как вы можете видеть сами, вы будете в конечном итоге с 4-х этапов, так как есть 3 зависимости перетасовать (ребра с номерами разделов).

Номера в круглых скобках - это количество разделов, которые DAGScheduler в конечном итоге будет использовать для создания наборов задач с точным количеством задач. Один TaskSet за этап.