Spark разделение на узлах foreachpartition - PullRequest
Новая
       26

Spark разделение на узлах foreachpartition

1 голос
/ 26 июня 2019

У меня есть спарк-кластер (DataProc) с мастером и 4-мя рабочими (2 предопределенных), в моем коде есть что-то вроде этого: 

    JavaRDD<Signal> rdd_data = javaSparkContext.parallelize(myArray);
rdd_data.foreachPartition(partitionOfRecords -> {
        while (partitionOfRecords.hasNext()) {
            MyData d = partitionOfRecords.next();
            LOG.info("my data: " + d.getId().toString());
        }
    })

myArray состоит из 1200 объектов MyData. Я не понимаю, почему в spark используются только 2 ядра, я делю массив на 2 раздела и не использую 16 ядер. Мне нужно установить номер раздела?

Заранее спасибо за любую помощь.

1 Ответ

1 голос
/ 26 июня 2019

Как правило, всегда полезно указывать число разделов в качестве второго аргумента для parallelize, поскольку оптимальное разбиение вашего набора данных должно быть действительно независимым от конкретной формы кластера, который вы используете, и Spark можетлучше всего использовать текущие размеры исполнителей в качестве «подсказки».

Здесь вы видите, что Spark по умолчанию запрашивает taskScheduler для текущего числа ядер исполнителей для использования в качествеdefaultParallelism, в сочетании с тем, что в Dataproc Spark динамическое выделение включено.Динамическое распределение важно, потому что в противном случае отдельное задание, отправленное в кластер, может просто указать максимальное количество исполнителей, даже если оно простаивает, и тогда это не позволит другим заданиям использовать эти незанятые ресурсы.

Так что в Dataproc, есливы используете по умолчанию n1-standard-4, Dataproc настраивает 2 исполнителя на каждую машину и дает каждому исполнителю 2 ядра.Значение spark.dynamicAllocation.minExecutors должно быть 1, поэтому задание по умолчанию при запуске без выполнения каких-либо работ будет выполняться на 1 исполнителе с 2 ядрами.Затем taskScheduler сообщит, что в настоящее время всего зарезервировано 2 ядра, и поэтому defaultParallelism будет равен 2.

Если у вас был большой кластер, и вы уже некоторое время выполняли задание (скажем, у вас есть картафаза, которая длится дольше 60 секунд) можно ожидать, что динамическое выделение займет все доступные ресурсы, поэтому следующим шагом задания, использующим defaultParallelism, будет, вероятно, 16, то есть общее количество ядер в вашем кластере (или, возможно, 14, если 2 используется приложением appmaster.)

На практике вы, вероятно, захотите распараллелить на большее количество разделов, чем общее количество доступных ядер в любом случае.Затем, если есть какой-то перекос в том, сколько времени занимает обработка каждого элемента, вы можете получить хорошую балансировку, когда быстрые задачи заканчиваются, и тогда эти исполнители могут начать брать новые разделы, пока медленные еще работают, вместо того, чтобы всегда ждать одногосамый медленный раздел до конца.Распространено выбирать количество разделов в любом месте от 2x числа доступных ядер до 100x или более.

Вот еще один связанный вопрос StackOverflow: spark.default.parallelism для Parallelize RDD по умолчанию равен 2 для sparkпредставить

...