Почему мое приложение Spark работает только у одного исполнителя?

Question

Я все еще довольно новичок в Spark, но мне удалось создать приложение Spark, которое мне нужно для обработки данных с нашего SQL Server с использованием драйверов JDBC (мы удаляем дорогостоящие SP), приложение загружает несколько таблиц из Sql Server через JDBC в dataframes, то я делаю несколько объединений, группу и фильтр, которые, наконец, повторно вставляют некоторые данные обратно через JDBC, результаты в другую таблицу. Все это отлично работает на Spark EMR в Amazon Web Services в m3.xlarge с 2 ядрами за минуту.

Мой вопрос: 1. прямо сейчас у меня есть 1 мастер и 2 ядра на кластере, но каждый раз, когда я запускаю новый шаг, кажется, из того, что я вижу на сервере истории, только 1 исполнитель используется, поскольку я могу видеть 2 перечисленных исполнителя, драйвер без использования вообще, исполнитель с обработкой id 1 вокруг 1410 задач. И я совершенно не уверен, как действовать дальше.

Также это касается AWS, но я не хотел публиковать 2 вопроса, поскольку они каким-то образом связаны, есть ли способ, которым я могу выполнить 2 шага одновременно? что означает, что у вас есть 2 искровых представления этого процесса, работающих одновременно, поскольку мы запускаем этот процесс много раз в день (он обрабатывает данные клиента). Я знаю, что я могу запустить новый кластер с шагом, но я хочу, чтобы иметь возможность быстро выполнять обработку, и просто запуск нового кластера занимает слишком много времени. Спасибо !!!

Answer 1

Для вашего первого вопроса:

Я не уверен, если это так, но что-то подобное случилось с нами, и, возможно, это может помочь.

Если вы читаете источник JDBC с использованием sqlContext.read.format("jdbc").load() (или аналогичного), по умолчанию результирующий фреймворк не секционируется. Итак, если это так, применение преобразований в результирующем фреймворке без его разбиения сначала приведет к тому, что только один исполнитель сможет его обработать. Если это не ваше дело, следующее решение, вероятно, не решит вашу проблему.

Итак, наше решение состояло в создании числового столбца со значениями от 1 до 32 (наше желаемое количество разделов) в данных и использование его в качестве разделительного столбца путем установки параметров разбиения на читательский блок jdbc (пожалуйста, отметьте this link):

val connectionOptions = Map[String, String] (... <connection options> ...) 
val options = connectionOptions ++ Map[String, String] (
    "partitionColumn" -> "column name", 
    "lowerBound" -> "1", 
    "upperBound" -> "32", 
    "numPartitions" -> "32" 
) 

val df = sqlContext.read.format("jdbc").options(options).load() 

Таким образом, при таком подходе не только была задача чтения в состоянии быть обработаны параллельно (на самом деле повышения производительности и избежать ошибки ООМ), но в результате dataframe распределяли и обрабатываться параллельно для всех последующие преобразования.

Я надеюсь, что это поможет.