Как я могу настроить из кода Java (или Scala) количество исполнителей, имеющих SparkConfig и SparkContext? Я постоянно вижу 2 исполнителя. Похоже, spark.default.parallelism не работает и что-то другое.
Мне просто нужно установить количество исполнителей равным размеру кластера, но всегда есть только два из них. Я знаю размер моего кластера. Я работаю на YARN, если это имеет значение.
ОК, понял.
Количество исполнителей на самом деле не является собственностью Spark, а скорее драйвером, используемым для размещения задания на YARN. Так как я использую класс SparkSubmit как драйвер, и у него есть соответствующий параметр --num-executors, который именно то, что мне нужно.
UPDATE:
Для некоторых заданий я больше не следую методу SparkSubmit. Я не могу сделать это в первую очередь для приложений, где Spark-задание является только одним из компонентов приложения (и даже необязательно). Для этих случаев я использую spark-defaults.conf, прикрепленный к конфигурации кластера, и свойство spark.executor.instances внутри него. Этот подход гораздо более универсален, что позволяет мне правильно сбалансировать ресурсы в зависимости от кластера (рабочей станции разработчика, постановки, производства).
Вы также можете сделать это программно, установив параметры "spark.executor.instances" и "spark.executor.cores" на объект SparkConf.
Пример:
SparkConf conf = new SparkConf()
// 4 workers
.set("spark.executor.instances", "4")
// 5 cores on each workers
.set("spark.executor.cores", "5");
Второй параметр предназначен только для YARN и автономного режима. Он позволяет приложению запускать несколько исполнителей на одного и того же работника при условии, что на этом рабочем месте достаточно ядер.
В Spark 2. 0+ версия
используйте переменную сеанса spark для динамического задания количества исполнителей (изнутри программы)
spark.conf.set("spark.executor.instances", 4)
spark.conf.set("spark.executor.cores", 4)
В вышеупомянутом случае максимум 16 задач будут выполнены в любой момент времени.
Другой вариант - динамическое распределение исполнителей, как показано ниже:
spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.executor.cores", 4)
spark.conf.set( "spark.dynamicAllocation.minExecutors", "1")
spark.conf.set( "spark.dynamicAllocation.maxExecutors", "5")
Таким образом, вы можете позволить spark принять решение о распределении числа исполнителей на основе требований к обработке и памяти для выполнения задания.
Я чувствую, что второй вариант работает лучше, чем первый и широко используется.
Надеюсь, это поможет.
У нас была похожая проблема в моей лаборатории при запуске Spark на Yarn с данными в формате hdf, но независимо от того, какое из приведенных выше решений я пробовал, я не мог увеличить число исполнителей Spark больше двух.
Оказывается, набор данных был слишком маленьким (меньше, чем размер блока hdfs в 128 МБ) и существовал только на двух узлах данных (1 мастер, 7 узлов данных в моем кластере) из-за эвристики репликации данных по умолчанию в hadoop.
Когда у меня и моих коллег по работе было больше файлов (и файлов большего размера), и данные были распределены по всем узлам, мы могли установить количество исполнителей Spark и, наконец, увидеть обратную зависимость между --num-executors и временем до завершения.
Надеюсь, что это помогает кому-то еще в аналогичной ситуации.