Запуск и управление приложениями для Spark и PySpark

Существует несколько способов запустить Spark- и PySpark-задания в кластере Yandex Data Processing:

• Spark Shell (командная оболочка для языков программирования Scala и Python). Расчеты запускаются не с помощью скрипта, а построчно. Подробнее о Spark Shell читайте в документации Spark.
• Скрипт spark-submit. Сохраняет результаты расчета в HDFS. Подробнее о spark-submit читайте в документации Spark.
• Команды CLI Yandex Cloud. Позволяют сохранить результаты расчета не только в HDFS, но и в бакете Yandex Object Storage.

Ниже рассматривается пример, по которому будет рассчитана статистика по воздушному трафику США за 2018 год по данным с сайта transtats.bts.gov. Набор данных представлен в формате Parquet и находится в публичном бакете Yandex Object Storage с именем yc-mdb-examples.

Перед началом работыПеред началом работы

Подготовьте инфраструктуру:

Использование Spark ShellИспользование Spark Shell

1. Подключитесь по SSH к хосту-мастеру кластера Yandex Data Processing.

2. Запустите Spark Shell на хосте-мастере:

   /usr/bin/pyspark
   

   Количество ядер и процессов выполнения задач (executor) ограничено только конфигурацией вашего кластера Yandex Data Processing.

3. Построчно введите следующий код:

   sql = SQLContext(sc)
   df = sql.read.parquet("s3a://yc-mdb-examples/dataproc/example01/set01")
   

   Последняя строка — чтение данных из публичного бакета с набором данных для примера. После выполнения этой строки в текущей сессии будет доступен организованный набор данных (DataFrame) df с прочитанными данными.

4. Чтобы увидеть схему полученного DataFrame, выполните команду:

   df.printSchema()
   

   В терминале будет выведен список столбцов с их типами.

5. Рассчитайте статистику перелетов по месяцам и найдите первую десятку городов по количеству вылетов:

   • Количество перелетов по месяцам:

     df.groupBy("Month").count().orderBy("Month").show()
     

   • Первая десятка городов по количеству вылетов:

     df.groupBy("OriginCityName").count().orderBy("count", ascending=False).show(10)
     

Использование Spark SubmitИспользование Spark Submit

Spark Submit позволяет запускать заранее написанные приложения через скрипт spark-submit. Для примера рассчитывается количество перелетов по месяцам.

Завершение работы приложенияЗавершение работы приложения

По умолчанию ресурсы запускаемого приложения управляются компонентом YARN. Если приложение необходимо завершить или убрать из очереди, используйте утилиту yarn:

1. Выведите список приложений:

   yarn application -list
   

2. Завершите ненужное приложение:

   yarn application -kill <идентификатор_приложения>
   

Более подробно с командами YARN можно ознакомиться на странице YARN Commands.

Запуск заданий (jobs) с помощью CLI Yandex CloudЗапуск заданий (jobs) с помощью CLI Yandex Cloud

Запуск заданий с помощью CLI Yandex Cloud происходит посредством агента Yandex Data Processing, установленного на хосте-мастере кластера. Параметры заданий передаются агенту через Yandex Data Processing API.

Исполняемый файл и его зависимости должны находиться в хранилище, к которому есть доступ у сервисного аккаунта кластера Yandex Data Processing. У самого запускаемого приложения должен быть доступ к хранилищам, в которых хранятся исходный набор данных и результаты запуска.

Результаты расчета можно сохранить в HDFS в кластере Yandex Data Processing или в бакете data-proc-bucket, указанном при создании кластера.

Служебная и отладочная информация сохраняется в бакете data-proc-bucket. Для каждого задания агент Yandex Data Processing создает отдельную папку с путем вида dataproc/clusters/<идентификатор_кластера>/jobs/<идентификатор_задачи>.

Ниже приведены два варианта приложения — для Python и Scala.

Запуск PySpark-заданияЗапуск PySpark-задания

Чтобы запустить PySpark-задание:

1. Установите дополнительные зависимости.
2. Подготовьте и запустите PySpark-задание.

Установите дополнительные зависимостиУстановите дополнительные зависимости

На локальном компьютере выполните действия:

1. Если у вас еще нет интерфейса командной строки Yandex Cloud, установите и инициализируйте его.

   По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

2. Установите и настройте консольный клиент S3cmd для работы с Yandex Object Storage.

3. Установите Python. Убедитесь, что версия Python совпадает с версией, доступной из образа. Проверить версию можно в разделе Среда исполнения. Для версии образа 2.0 используйте Python 3.8.10:

   sudo apt update && sudo apt install python3.8
   

Подготовьте и запустите PySpark-заданиеПодготовьте и запустите PySpark-задание

1. Создайте файл job.py со следующим кодом:

   import sys
   
   from pyspark import SparkContext, SparkConf
   from pyspark.sql import SQLContext
   
   def main():
   
       if len(sys.argv) != 3:
           print('Usage job.py <входная_директория> <выходная_директория>')
           sys.exit(1)
   
       in_dir = sys.argv[1]
       out_dir = sys.argv[2]
   
       conf = SparkConf().setAppName('Month Stat - Python')
       sc = SparkContext(conf=conf)
       sql = SQLContext(sc)
       df = sql.read.parquet(in_dir)
       month_stat = df.groupBy('Month').count()
       job_id = dict(sc._conf.getAll())['spark.yarn.tags'].replace('dataproc_job_', '')
       if out_dir.startswith('s3a://'):
           month_stat.repartition(1).write.format('csv').save(out_dir + job_id)
       else:
           default_fs = sc._jsc.hadoopConfiguration().get('fs.defaultFS')
           month_stat.repartition(1).write.format('csv').save(default_fs + out_dir + job_id)
   
   
   if __name__ == '__main__':
       main()
   

2. Чтобы PySpark имел доступ к вашему коду, загрузите файл job.py в бакет Object Storage, к которому есть доступ у сервисного аккаунта кластера Yandex Data Processing:

   s3cmd put ./job.py s3://data-proc-bucket/bin/
   

3. Запустите задание.

   Команда для запуска зависит от того, где нужно сохранить результаты задания: в Object Storage или HDFS.

   Object Storage

   Директория HDFS

   yc dataproc job create-pyspark \
      --cluster-id=<идентификатор_кластера> \
      --name=<имя_задачи> \
      --main-python-file-uri="s3a://data-proc-bucket/bin/job.py" \
      --args="s3a://yc-mdb-examples/dataproc/example01/set01" \
      --args="s3a://data-proc-bucket/jobs_results/"
   

   В команде укажите:

   • --cluster-id — идентификатор кластера. Его можно получить со списком кластеров в каталоге.
   • --name — произвольное имя Spark-задания.

   CSV-файл с результатом сохранится в бакете data-proc-bucket.

   yc dataproc job create-pyspark \
      --cluster-id=<идентификатор_кластера> \
      --name=<имя_задачи> \
      --main-python-file-uri="s3a://data-proc-bucket/bin/job.py" \
      --args="s3a://yc-mdb-examples/dataproc/example01/set01" \
      --args="tmp/jobs/"
   

   В команде укажите:

   • --cluster-id — идентификатор кластера. Его можно получить со списком кластеров в каталоге.
   • --name — произвольное имя Spark-задания.

   CSV-файл с результатом сохранится в папке /tmp/jobs/<идентификатор_задачи>/ в HDFS.

4. (Опционально) Посмотрите логи задачи:

   yc dataproc job log <идентификатор_задачи> --cluster-id=<идентификатор_кластера>
   

Запуск Spark-заданияЗапуск Spark-задания

Чтобы запустить Spark-задание:

1. Установите дополнительные зависимости.
2. Соберите Scala-приложение.
3. Загрузите JAR-файл в Object Storage.
4. Запустите Spark-задание в кластере Yandex Data Processing.

Установите дополнительные зависимостиУстановите дополнительные зависимости

1. Если у вас еще нет интерфейса командной строки Yandex Cloud, установите и инициализируйте его.

   По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

2. Подключитесь по SSH к хосту-мастеру кластера Yandex Data Processing.

3. Установите стандартную утилиту сборки sbt для Scala. Она устанавливается вместе с языком программирования Scala.

4. Установите и настройте консольный клиент S3cmd для работы с Yandex Object Storage.

Соберите Scala-приложениеСоберите Scala-приложение

Чтобы упростить управление зависимостями, соберите приложение в один JAR-файл (fat JAR) с помощью плагина sbt-assembly:

1. Создайте папку spark-app, в ней — папки project и src/main/scala.

2. Создайте файл spark-app/project/plugins.sbt, который описывает подключение плагина sbt-assembly для сборки единого JAR-файла:

   addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "<версия_sbt-assembly>")
   

   Версию плагина sbt-assembly см. в его репозитории, в разделе Releases.

3. Узнайте установленную версию Scala с помощью команды scala -version.

4. Создайте файл spark-app/build.sbt с описанием зависимостей и стратегии их слияния в одном JAR-файле. В файле build.sbt укажите версию Scala:

   scalaVersion := "<версия_Scala>"
   
   libraryDependencies ++= Seq(
       "org.apache.spark" %% "spark-core" % "2.4.4",
       "org.apache.spark" %% "spark-sql" % "2.4.4",
   )
   
   assembly / assemblyMergeStrategy := {
     case PathList("org","aopalliance", xs @ _*) => MergeStrategy.last
     case PathList("javax", "inject", xs @ _*) => MergeStrategy.last
     case PathList("javax", "servlet", xs @ _*) => MergeStrategy.last
     case PathList("javax", "activation", xs @ _*) => MergeStrategy.last
     case PathList("org", "apache", xs @ _*) => MergeStrategy.last
     case PathList("com", "google", xs @ _*) => MergeStrategy.last
     case PathList("com", "esotericsoftware", xs @ _*) => MergeStrategy.last
     case PathList("com", "codahale", xs @ _*) => MergeStrategy.last
     case PathList("com", "yammer", xs @ _*) => MergeStrategy.last
     case "about.html" => MergeStrategy.rename
     case "overview.html" => MergeStrategy.last
     case "META-INF/ECLIPSEF.RSA" => MergeStrategy.last
     case "META-INF/mailcap" => MergeStrategy.last
     case "META-INF/mimetypes.default" => MergeStrategy.last
     case "plugin.properties" => MergeStrategy.last
     case "log4j.properties" => MergeStrategy.last
     case "git.properties" => MergeStrategy.last
     case x =>
       val oldStrategy = (assembly / assemblyMergeStrategy).value
       oldStrategy(x)
   }
   

5. Создайте файл spark-app/src/main/scala/app.scala с кодом приложения:

   package com.yandex.cloud.dataproc.scala
   
   import org.apache.spark.{SparkConf, SparkContext}
   import org.apache.spark.sql.SQLContext
   
     object Main {
       def main(args: Array[String]) {
         if (args.length != 2){ //проверяем аргумент
           System.err.println("Usage spark-app.jar <входная_директория> <выходная_директория>");
           System.exit(-1);
         }
         val inDir = args(0); //URI на исходные данные
         val outDir = args(1); //URI на директорию, куда записать результат
         val conf = new SparkConf().setAppName("Month Stat - Scala App")
         val sc = new SparkContext(conf)
         val sqlContext = new org.apache.spark.sql.SQLContext(sc)
         val df = sqlContext.read.parquet(inDir)
         val monthStat = df.groupBy("Month").count()
         val defaultFS = sc.hadoopConfiguration.get("fs.defaultFS") //получить эндпоинт сервера HDFS
         val jobId = conf.get("spark.yarn.tags").replace("dataproc_job_", ""); //получить идентификатор задания
         if (outDir.toLowerCase().startsWith("s3a://")) {
           monthStat.repartition(1).write.format("csv").save(outDir + jobId)
         } else {
           monthStat.repartition(1).write.format("csv").save(defaultFS + "/" + outDir + jobId)
         }
   
         sc.stop()
       }
     }
   

6. Запустите сборку приложения в папке spark-app:

   sbt clean && sbt compile && sbt assembly
   

   В случае ошибки Error looking up function 'stat'

   Если вы получили ошибку java.lang.UnsatisfiedLinkError: Error looking up function 'stat': java: undefined symbol: stat и ОС хоста-мастера — Ubuntu, запустите каждую команду sbt с флагом -Dsbt.io.jdktimestamps=true:

   sbt clean -Dsbt.io.jdktimestamps=true && \
   sbt compile -Dsbt.io.jdktimestamps=true && \
   sbt assembly -Dsbt.io.jdktimestamps=true
   

Файл будет доступен по следующему пути: spark-app/target/scala-<версия_Scala>/spark-app-assembly-0.1.0-SNAPSHOT.jar.

Загрузите JAR-файл в Object StorageЗагрузите JAR-файл в Object Storage

Чтобы Spark имел доступ к собранному JAR-файлу, загрузите файл в бакет data-proc-bucket. Загрузить файл можно с помощью s3cmd:

s3cmd put ~/spark-app/target/scala-<версия_Scala>/spark-app-assembly-0.1.0-SNAPSHOT.jar s3://data-proc-bucket/bin/

Файл загружается по адресу s3://data-proc-bucket/bin/spark-app-assembly-0.1.0-SNAPSHOT.jar.

Запустите Spark-задание в кластере Yandex Data ProcessingЗапустите Spark-задание в кластере Yandex Data Processing

1. Отключитесь от хоста-мастера кластера.

2. Запустите задание.

   Команда для запуска зависит от того, где нужно сохранить результаты задания: в Object Storage или HDFS.

   Object Storage

   Директория HDFS

   yc dataproc job create-spark \
      --cluster-id=<идентификатор_кластера> \
      --name=<имя_задачи> \
      --main-class="com.yandex.cloud.dataproc.scala.Main" \
      --main-jar-file-uri="s3a://data-proc-bucket/bin/spark-app-assembly-0.1.0-SNAPSHOT.jar" \
      --args="s3a://yc-mdb-examples/dataproc/example01/set01" \
      --args="s3a://data-proc-bucket/jobs_results/"
   

   В команде укажите:

   • --cluster-id — идентификатор кластера. Его можно получить со списком кластеров в каталоге.
   • --name — произвольное имя Spark-задания.

   CSV-файл с результатом сохранится в бакете data-proc-bucket.

   yc dataproc job create-spark \
      --cluster-id=<идентификатор_кластера> \
      --name=<имя_задачи> \
      --main-class="com.yandex.cloud.dataproc.scala.Main" \
      --main-jar-file-uri="s3a://data-proc-bucket/bin/spark-app-assembly-0.1.0-SNAPSHOT.jar" \
      --args="s3a://yc-mdb-examples/dataproc/example01/set01" \
      --args="tmp/jobs/"
   

   В команде укажите:

   • --cluster-id — идентификатор кластера. Его можно получить со списком кластеров в каталоге.
   • --name — произвольное имя Spark-задания.

   CSV-файл с результатом сохранится в папке /tmp/jobs/<идентификатор_задачи>/ в HDFS.

   Пример сообщения об успешном запуске задачи:

   done (1m2s)
   id: {your_job_id}
   cluster_id: {your_cluster_id}
   name: test02
   status: DONE
   spark_job:
     args:
     - s3a://yc-mdb-examples/dataproc/example01/set01
     - s3a://data-proc-bucket/jobs_results/
     main_jar_file_uri: s3a://data-proc-bucket/bin/spark-app-assembly-0.1.0-SNAPSHOT.jar
     main_class: com.yandex.cloud.dataproc.scala.Main
   

Удалите созданные ресурсыУдалите созданные ресурсы

Некоторые ресурсы платные. Чтобы за них не списывалась плата, удалите ресурсы, которые вы больше не будете использовать: