Apache Spark - Развертывание

Приложение Spark, использующее spark-submit, представляет собой команду оболочки, используемую для развертывания приложения Spark в кластере. Он использует все соответствующие менеджеры кластера через единый интерфейс. Таким образом, вам не нужно настраивать приложение для каждого из них.

пример

Давайте возьмем тот же пример подсчета слов, который мы использовали ранее, с использованием команд оболочки. Здесь мы рассматриваем тот же пример, что и искровое приложение.

Пример ввода

Следующий текст - это входные данные, а файл с именем in.txt.

people are not as beautiful as they look, 
as they walk or as they talk. 
they are only as beautiful  as they love, 
as they care as they share.

Посмотрите на следующую программу -

SparkWordCount.scala

import org.apache.spark.SparkContext 
import org.apache.spark.SparkContext._ 
import org.apache.spark._  

object SparkWordCount { 
   def main(args: Array[String]) { 

      val sc = new SparkContext( "local", "Word Count", "/usr/local/spark", Nil, Map(), Map()) 
		
      /* local = master URL; Word Count = application name; */  
      /* /usr/local/spark = Spark Home; Nil = jars; Map = environment */ 
      /* Map = variables to work nodes */ 
      /*creating an inputRDD to read text file (in.txt) through Spark context*/ 
      val input = sc.textFile("in.txt") 
      /* Transform the inputRDD into countRDD */ 
		
      val count = input.flatMap(line ⇒ line.split(" ")) 
      .map(word ⇒ (word, 1)) 
      .reduceByKey(_ + _) 
       
      /* saveAsTextFile method is an action that effects on the RDD */  
      count.saveAsTextFile("outfile") 
      System.out.println("OK"); 
   } 
}

Сохраните указанную выше программу в файл с именем SparkWordCount.scala и поместите его в определенный пользователем каталог с именем spark-application.

Note - При преобразовании inputRDD в countRDD мы используем flatMap () для токенизации строк (из текстового файла) в слова, метод map () для подсчета частоты слов и метод reduceByKey () для подсчета повторения каждого слова.

Выполните следующие действия, чтобы отправить это приложение. Выполните все шаги вspark-application каталог через терминал.

Шаг 1. Загрузите Spark Ja

Для компиляции требуется Spark core jar, поэтому загрузите spark-core_2.10-1.3.0.jar по следующей ссылке Spark core jar и переместите файл jar из каталога загрузки вspark-application каталог.

Шаг 2: Скомпилируйте программу

Скомпилируйте указанную выше программу, используя приведенную ниже команду. Эта команда должна выполняться из каталога приложения spark. Вот,/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar - это банка поддержки Hadoop, взятая из библиотеки Spark.

$ scalac -classpath "spark-core_2.10-1.3.0.jar:/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar" SparkPi.scala

Шаг 3. Создайте JAR

Создайте файл jar приложения Spark, используя следующую команду. Вот,wordcount это имя файла для файла jar.

jar -cvf wordcount.jar SparkWordCount*.class spark-core_2.10-1.3.0.jar/usr/local/spark/lib/spark-assembly-1.4.0-hadoop2.6.0.jar

Шаг 4. Подайте заявку на искру

Отправьте приложение Spark, используя следующую команду -

spark-submit --class SparkWordCount --master local wordcount.jar

Если он выполнен успешно, вы найдете вывод, приведенный ниже. ВOKввод следующего вывода предназначен для идентификации пользователя, и это последняя строка программы. Если вы внимательно прочитаете следующий вывод, вы найдете разные вещи, такие как -

успешно запустил службу sparkDriver на порту 42954
MemoryStore запущен с объемом 267,3 МБ
Запустил SparkUI на http://192.168.1.217:4040
Добавлен файл JAR: /home/hadoop/piapplication/count.jar
ResultStage 1 (saveAsTextFile на SparkPi.scala: 11) завершился за 0,566 с.
Остановлен веб-интерфейс Spark на http://192.168.1.217:4040
MemoryStore очищен

15/07/08 13:56:04 INFO Slf4jLogger: Slf4jLogger started 
15/07/08 13:56:04 INFO Utils: Successfully started service 'sparkDriver' on port 42954. 
15/07/08 13:56:04 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://[email protected]:42954] 
15/07/08 13:56:04 INFO MemoryStore: MemoryStore started with capacity 267.3 MB 
15/07/08 13:56:05 INFO HttpServer: Starting HTTP Server 
15/07/08 13:56:05 INFO Utils: Successfully started service 'HTTP file server' on port 56707. 
15/07/08 13:56:06 INFO SparkUI: Started SparkUI at http://192.168.1.217:4040 
15/07/08 13:56:07 INFO SparkContext: Added JAR file:/home/hadoop/piapplication/count.jar at http://192.168.1.217:56707/jars/count.jar with timestamp 1436343967029 
15/07/08 13:56:11 INFO Executor: Adding file:/tmp/spark-45a07b83-42ed-42b3b2c2-823d8d99c5af/userFiles-df4f4c20-a368-4cdd-a2a7-39ed45eb30cf/count.jar to class loader 
15/07/08 13:56:11 INFO HadoopRDD: Input split: file:/home/hadoop/piapplication/in.txt:0+54 
15/07/08 13:56:12 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 2001 bytes result sent to driver 
 (MapPartitionsRDD[5] at saveAsTextFile at SparkPi.scala:11), which is now runnable 
15/07/08 13:56:12 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 1 (MapPartitionsRDD[5] at saveAsTextFile at SparkPi.scala:11) 
15/07/08 13:56:13 INFO DAGScheduler: ResultStage 1 (saveAsTextFile at SparkPi.scala:11) finished in 0.566 s 
15/07/08 13:56:13 INFO DAGScheduler: Job 0 finished: saveAsTextFile at SparkPi.scala:11, took 2.892996 s
OK 
15/07/08 13:56:13 INFO SparkContext: Invoking stop() from shutdown hook 
15/07/08 13:56:13 INFO SparkUI: Stopped Spark web UI at http://192.168.1.217:4040 
15/07/08 13:56:13 INFO DAGScheduler: Stopping DAGScheduler 
15/07/08 13:56:14 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped! 
15/07/08 13:56:14 INFO Utils: path = /tmp/spark-45a07b83-42ed-42b3-b2c2823d8d99c5af/blockmgr-ccdda9e3-24f6-491b-b509-3d15a9e05818, already present as root for deletion. 
15/07/08 13:56:14 INFO MemoryStore: MemoryStore cleared 
15/07/08 13:56:14 INFO BlockManager: BlockManager stopped 
15/07/08 13:56:14 INFO BlockManagerMaster: BlockManagerMaster stopped 
15/07/08 13:56:14 INFO SparkContext: Successfully stopped SparkContext 
15/07/08 13:56:14 INFO Utils: Shutdown hook called 
15/07/08 13:56:14 INFO Utils: Deleting directory /tmp/spark-45a07b83-42ed-42b3b2c2-823d8d99c5af 
15/07/08 13:56:14 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!

Шаг 5: Проверка вывода

После успешного выполнения программы вы найдете каталог с именем outfile в каталоге Spark-application.

Следующие команды используются для открытия и проверки списка файлов в каталоге Outfile.

$ cd outfile 
$ ls 
Part-00000 part-00001 _SUCCESS

Команды для проверки вывода в part-00000 файл -

$ cat part-00000 
(people,1) 
(are,2) 
(not,1) 
(as,8) 
(beautiful,2) 
(they, 7) 
(look,1)

Команды для проверки вывода в файле part-00001:

$ cat part-00001 
(walk, 1) 
(or, 1) 
(talk, 1) 
(only, 1) 
(love, 1) 
(care, 1) 
(share, 1)

Просмотрите следующий раздел, чтобы узнать больше о команде 'spark-submit'.

Синтаксис Spark-submit

spark-submit [options] <app jar | python file> [app arguments]

Параметры

В таблице ниже описан список options -

S.No	Вариант	Описание
1	--мастер	spark: // хост: порт, mesos: // хост: порт, пряжа или локальный.
2	--deploy-mode	Следует ли запускать программу драйвера локально («клиент») или на одной из рабочих машин внутри кластера («кластер») (по умолчанию: клиент).
3	--класс	Основной класс вашего приложения (для приложений Java / Scala).
4	--название	Имя вашего приложения.
5	- банки	Разделенный запятыми список локальных jar-файлов для включения в пути к классам драйвера и исполнителя.
6	--пакеты	Разделенный запятыми список maven-координат jar-файлов для включения в пути к классам драйвера и исполнителя.
7	- репозитории	Разделенный запятыми список дополнительных удаленных репозиториев для поиска координат maven, заданных с помощью --packages.
8	--py-файлы	Разделенный запятыми список файлов .zip, .egg или .py для размещения в PYTHON PATH для приложений Python.
9	--файлы	Разделенный запятыми список файлов, которые необходимо разместить в рабочем каталоге каждого исполнителя.
10	--conf (prop = val)	Произвольное свойство конфигурации Spark.
11	--properties-файл	Путь к файлу, из которого можно загрузить дополнительные свойства. Если не указано иное, будет выполняться поиск параметров conf / spark-defaults.
12	--driver-memory	Память для драйвера (например, 1000 МБ, 2 ГБ) (по умолчанию: 512 МБ).
13	--driver-java-options	Дополнительные параметры Java для передачи драйверу.
14	- путь-библиотека-драйвер	Дополнительные записи пути к библиотеке для передачи драйверу.
15	--driver-class-path	Дополнительные записи пути к классу для передачи драйверу. Обратите внимание, что банки, добавленные с помощью --jars, автоматически включаются в путь к классам.
16	--executor-memory	Память на исполнителя (например, 1000 МБ, 2 ГБ) (по умолчанию: 1 ГБ).
17	--proxy-пользователь	Пользователь, который будет выдавать себя за другое лицо при подаче заявки.
18	--help, -h	Показать это справочное сообщение и выйти.
19	--verbose, -v	Распечатать дополнительный вывод отладки.
20	--версия	Распечатайте версию текущего Spark.
21 год	--driver-cores ЧИСЛО	Ядра для драйвера (по умолчанию: 1).
22	- контролировать	Если задано, перезапускает драйвер в случае ошибки.
23	--убийство	Если задано, убивает указанный драйвер.
24	--положение дел	Если задано, запрашивает статус указанного драйвера.
25	--total-executor-cores	Всего ядер для всех исполнителей.
26	--executor-cores	Количество ядер на исполнителя. (По умолчанию: 1 в режиме YARN или все доступные ядра на рабочем сервере в автономном режиме).