Logging Spark Jobs

Question

Я пытаюсь отслеживать задания, которые отправляются в кластер, и до сих пор находил решения для регистрации журналов событий, используя spark.eventLog.enabled = True, который предоставляет информацию о том, когда начинаются задачи и (подробнее об этом here) или log4j, который также предоставляет информацию о состоянии и ходе выполнения задачи.

Что я действительно хочу, так это зарегистрировать задачу, которая запускается. Таким образом, это захватит код, который был выполнен следующим образом: var = sc.range (1000) или min_var = var.min(). Из того, что я видел, другие объяснители, описанные выше, не могут этого сделать.

В качестве примера, если я пробежал две команды выше (вар = sc.range (1000) и min_var = var.min()) Я хотел бы увидеть что-то вроде следующего в log4j- как регистратор:

INFO RUNNING переменная = sc.range (1000)

INFO RUNNING min_var = var.min()

кто-нибудь натыкался регистратору, как это?

Answer 1

Если вы работаете на пряжу и yarn.log-aggregation-enable установлен в true вы могли бы сделать:

yarn logs --applicationId <application-id> 

и получить журналы для любого конечного применения, в том числе журналов, которые были созданы с помощью кода.

---

UPDATE:

К сожалению, там нет (по крайней мере, популярной) библиотека, которая позволяет вам сделать это (занесен в журнал в-между-постановщиков границами кода), но в остальном вы можете использовать искру журналы водителя, чтобы получить что-то более учебное, чем то, что у вас есть сейчас. Сначала убедитесь, что вы получите максимальную отдачу от бревен путем установки log4j уровня DEBUG (создать из шаблона и редактировать conf/log4j.properties) следующим образом:

log4j.rootCategory=DEBUG, console 

Затем, журналы делают некоторую фильтрацию (особенно на журналы несуществующей водителя). Например, для работы:

user@laptop:~$ cd ~/opt/spark 
user@laptop:~/opt/spark$ git clone https://github.com/ehiggs/spark-terasort.git 
user@laptop:~/opt/spark$ cd spark-terasort 
user@laptop:~/opt/spark/spark-terasort$ mvn package 
... 
user@laptop:~/opt/spark/spark-terasort$ cd .. 
user@laptop:~/opt/spark$ ./bin/spark-submit --class com.github.ehiggs.spark.terasort.TeraSort spark-terasort/target/spark-terasort-1.0-SNAPSHOT-jar-with-dependencies.jar ~/data/terasort_in ~/data/terasort_out &> logs.logs 

Тогда вы могли бы сделать:

user@laptop:~/opt/spark$ cat logs.logs | grep "Registering RDD\|Final stage\|Job" | grep DAG 

и получить что-то вроде:

16/03/24 23:47:44 INFO DAGScheduler: Registering RDD 0 (newAPIHadoopFile at TeraSort.scala:60) 
16/03/24 23:47:44 INFO DAGScheduler: Final stage: ResultStage 1 (sortByKey at TeraSort.scala:61) 
16/03/24 23:48:41 INFO DAGScheduler: Job 0 finished: sortByKey at TeraSort.scala:61, took 56.468248 s 
16/03/24 23:48:41 INFO DAGScheduler: Registering RDD 1 (partitionBy at TeraSort.scala:61) 
16/03/24 23:48:41 INFO DAGScheduler: Final stage: ResultStage 4 (saveAsNewAPIHadoopFile at TeraSort.scala:62) 
16/03/24 23:50:35 INFO DAGScheduler: Job 1 finished: saveAsNewAPIHadoopFile at TeraSort.scala:62, took 114.042019 s 

Следует отметить, что узкие преобразования, которые не хвост из фаза не будет указана. Здесь, а не добавлять протоколирование вызовов в код, вы могли бы обогатить имена РДДА, делающее использование следующей уловки:

rdd.setName("more interesting info, or even the algorithm itself") 

и получить его отображается в Спарке журналов себя в качестве руководства.

Надеюсь, это даст вам некоторые идеи поближе к тому, что вы ожидаете.