Какая задача в Spark? Как рабочий Spark выполняет файл jar?

После прочтения некоторого документа на http://spark.apache.org/docs/0.8.0/cluster-overview.html у меня возник вопрос, который я хочу уточнить.

Возьмем этот пример из Spark:

JavaSparkContext spark = new JavaSparkContext(
  new SparkConf().setJars("...").setSparkHome....);
JavaRDD<String> file = spark.textFile("hdfs://...");

// step1
JavaRDD<String> words =
  file.flatMap(new FlatMapFunction<String, String>() {
    public Iterable<String> call(String s) {
      return Arrays.asList(s.split(" "));
    }
  });

// step2
JavaPairRDD<String, Integer> pairs =
  words.map(new PairFunction<String, String, Integer>() {
    public Tuple2<String, Integer> call(String s) {
      return new Tuple2<String, Integer>(s, 1);
    }
  });

// step3
JavaPairRDD<String, Integer> counts =
  pairs.reduceByKey(new Function2<Integer, Integer>() {
    public Integer call(Integer a, Integer b) {
      return a + b;
    }
  });

counts.saveAsTextFile("hdfs://...");

Итак, скажем, у меня есть кластер из 3 узлов, а node 1 работает как master, а вышеприведенная программа драйвера была правильно вписана (скажем, application-test.jar). Итак, теперь я запускаю этот код на master node, и я считаю, что сразу после создания SparkContext файл application-test.jar будет скопирован на рабочие узлы (и каждый работник создаст каталог для этого приложение).

Итак, теперь мой вопрос: Являются ли шаги step1, step2 и step3 в примерах задач, которые отправляются работникам? Если да, то как рабочий выполняет это? Как java -cp "application-test.jar" step1 и так далее?

Ответы

Ответ 1

При создании SparkContext каждый рабочий запускает исполнитель. Это отдельный процесс (JVM), и он также загружает вашу банку. Исполнители подключаются к вашей программе драйверов. Теперь драйвер может отправлять им команды, например flatMap, map и reduceByKey в вашем примере. Когда водитель завершает работу, исполнители завершают работу.

RDD похожи на большие массивы, разделенные на разделы, и каждый исполнитель может удерживать некоторые из этих разделов.

Задача - это команда, отправленная из драйвера исполнителю путем сериализации вашего объекта Function. Исполнитель выполняет десериализацию команды (это возможно, потому что она загрузила вашу банку) и выполняет ее на разделе.

(Это концептуальный обзор. Я замалчиваю некоторые детали, но надеюсь, что это будет полезно.)

Чтобы ответить на ваш конкретный вопрос: Нет, новый процесс не запускается для каждого шага. Новый процесс запускается для каждого работника при построении SparkContext.

Ответ 2

Чтобы получить четкое представление о том, как создаются и планируются задачи, мы должны понять, как работает модель исполнения в Spark. Вкратце, приложение из искры выполняется в три этапа:

  • Создать график RDD
  • Создать план выполнения в соответствии с графиком RDD. Этапы создаются на этом этапе.
  • Создание задач на основе плана и их планирование по рабочим группам

В примере с примером слов граф RDD довольно прост, он выглядит следующим образом:

file → lines → words → per-word count → глобальное число слов → вывод

На основе этого графика создаются два этапа. Правило создания сцены основано на идее конвейера как можно большего числа узких преобразований. В вашем примере узкое преобразование заканчивается подсчетом слов. Таким образом, вы получаете две стадии

  • file → lines → words → per-word count
  • глобальное число слов → вывод

После того, как этапы будут разобраны, искра будет генерировать задачи из этапов. На первом этапе будут созданы ShuffleMapTasks, и последний этап создаст ResultTasks, потому что на последнем этапе для выполнения результатов включается одна операция действия.

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

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

Как правило, количество заданий, выполняемых для этапа, - это точно количество разделов конечного RDD, но поскольку RDD могут быть разделены (и, следовательно, ShuffleMapStages), их количество варьируется в зависимости от совместного использования RDD/stage. См. Как DAG работает под обложками в RDD?