Как рассчитать наилучшее числоOfPartitions для объединения?

Итак, я понимаю, что в общем случае следует использовать coalesce() когда:

количество разделов уменьшается из-за filter или какой-либо другой операции, что может привести к уменьшению исходного набора данных (RDD, DF). coalesce() полезен для более эффективного выполнения операций после фильтрации большого набора данных.

Я также понимаю, что он менее дорогостоящий, чем repartition поскольку он уменьшает перетасовку, перемещая данные только при необходимости. Моя проблема заключается в том, как определить параметр, который выполняет coalesce (idealPartionionNo). Я работаю над проектом, который был передан мне от другого инженера, и он использовал приведенный ниже расчет, чтобы вычислить значение этого параметра.

// DEFINE OPTIMAL PARTITION NUMBER
implicit val NO_OF_EXECUTOR_INSTANCES = sc.getConf.getInt("spark.executor.instances", 5)
implicit val NO_OF_EXECUTOR_CORES = sc.getConf.getInt("spark.executor.cores", 2)

val idealPartionionNo = NO_OF_EXECUTOR_INSTANCES * NO_OF_EXECUTOR_CORES * REPARTITION_FACTOR

Затем он используется с объектом partitioner:

val partitioner = new HashPartitioner(idealPartionionNo)

но также используется с:

RDD.filter(x=>x._3<30).coalesce(idealPartionionNo)

Правильно ли это? В чем заключается основная идея вычисления idealPartionionNo? Что такое REPARTITION_FACTOR? Как я обычно это определяю?

Кроме того, поскольку YARN отвечает за определение доступных исполнителей "на лету", есть способ получить это число (AVAILABLE_EXECUTOR_INSTANCES) на лету и использовать его для вычисления idealPartionionNo (т.е. заменить NO_OF_EXECUTOR_INSTANCES на AVAILABLE_EXECUTOR_INSTANCES)?

В идеале, некоторые фактические примеры формы:

Вот набор данных (размер);
Здесь имеется ряд преобразований и возможных повторений RDD/DF.
Здесь вы должны перераспределить/объединить.
Предположим, что у вас есть n исполнителей с m ядрами и коэффициент разбиения, равный k

тогда:

Идеальное количество разделов будет ==>???

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

Ответ 1

На практике оптимальное количество разделов в большей степени зависит от данных, которые вы используете, от используемых преобразований и общей конфигурации, чем от доступных ресурсов.

Если количество разделов слишком низкое, вы будете испытывать длительные паузы GC, различные типы проблем с памятью и, наконец, субоптимальное использование ресурсов.
Если количество разделов слишком велико, то стоимость обслуживания может легко превысить стоимость обработки. Более того, если вы используете нераспределенные операции сокращения (например, reduce в отличие от treeReduce), большое количество разделов приводит к более высокой нагрузке на драйвер.

Вы можете найти ряд правил, которые предполагают переподписку разделов по сравнению с количеством ядер (фактор 2 или 3, кажется, общий) или хранения разделов определенного размера, но это не учитывает ваш собственный код:

Если вы выделяете много, вы можете ожидать длительные паузы GC, и, вероятно, лучше пойти с меньшими разделами.
Если какой-то фрагмент кода стоит дорого, ваши затраты в случайном порядке могут быть амортизированы более высоким уровнем параллелизма.
Если у вас есть фильтр, вы можете настроить количество разделов на основе дискриминирующей способности предиката (вы принимаете различные решения, если вы хотите сохранить 5% данных и 99% данных).

По моему мнению:

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

Обычно относительно просто определить количество необработанных данных на раздел, для которого ваш кластер демонстрирует стабильное поведение (по моему опыту он находится где-то в диапазоне нескольких сотен мегабайт, в зависимости от формата, структуры данных, которую вы используете для загрузки данных, и конфигурация). Это "волшебное число", которое вы ищете.

Некоторые вещи, которые вы должны помнить в целом:

Количество разделов не обязательно отражает распределение данных. Любая операция, требующая перетасовки (*byKey, join, RDD.partitionBy, Dataset.repartition), может привести к неравномерному распределению данных. Всегда следите за своими работами за симптомы значительного перекоса данных.
Количество разделов вообще не является постоянным. Любая операция с несколькими зависимостями (union, coGroup, join) может влиять на количество разделов.

Ответ 2

Ваш вопрос является допустимым, но оптимизация разбиения Spark полностью зависит от выполняемого вами вычисления. У вас должна быть веская причина для перераспределения/объединения; если вы просто считаете RDD (даже если у него огромное количество малонаселенных разделов), то любой шаг перераспределения/слияния просто замедлит вас.

Перегруппировка против coalesce

Разница между repartition(n) (которая совпадает с coalesce(n, shuffle = true) и coalesce(n, shuffle = false) связана с моделью исполнения. Модель Shuffle берет каждый раздел в исходном RDD, случайно отправляет его данные для всех исполнителей, и приводит к RDD с новым (меньшим или большим) количеством разделов. Модель no-shuffle создает новый RDD, который загружает несколько разделов в качестве одной задачи.

Рассмотрим это вычисление:

sc.textFile("massive_file.txt")
  .filter(sparseFilterFunction) // leaves only 0.1% of the lines
  .coalesce(numPartitions, shuffle = shuffle)

Если shuffle true, то вычисления текстовых файлов/фильтров происходят в ряде задач, заданных значениями по умолчанию в textFile, и крошечные отфильтрованные результаты перетасовываются. Если shuffle false, то количество заданий не более numPartitions.

Если numPartitions равно 1, то разница довольно суровая. Модель shuffle будет обрабатывать и фильтровать данные параллельно, а затем отправлять 0,1% отфильтрованных результатов одному исполнителю для последующих операций DAG. Модель no-shuffle будет обрабатывать и фильтровать данные на одном ядре с самого начала.

Действия

Рассмотрите свои последующие операции. Если вы просто используете этот набор данных один раз, то вам, вероятно, вообще не нужно переделки. Если вы сохраняете отфильтрованный RDD для последующего использования (например, на диске), рассмотрите вышеперечисленные компромиссы. Для ознакомления с этими моделями и опытом, когда вы работаете лучше, попробуйте оба и посмотрите, как они работают!

Ответ 3

Как ответили другие, нет формулы, которая вычисляет то, о чем вы просите. Тем не менее, вы можете сделать обоснованное предположение о первой части, а затем точно настроить ее с течением времени.

Первый шаг - убедиться, что у вас достаточно разделов. Если у вас есть исполнители NO_OF_EXECUTOR_INSTANCES и NO_OF_EXECUTOR_CORES для каждого исполнителя, вы можете обрабатывать разделы NO_OF_EXECUTOR_INSTANCES * NO_OF_EXECUTOR_CORES в одно и то же время (каждый из них будет идти в конкретное ядро конкретного экземпляра). Тем не менее, это предполагает, что все разделено поровну между ядрами, и все занимает ровно одно и то же время для обработки. Это редко бывает. Существует хорошая вероятность того, что некоторые из них будут закончены перед другими либо из-за локальности (например, данные должны поступать с другого узла), либо просто потому, что они не сбалансированы (например, если у вас есть данные, разделенные корневым доменом, тогда разделы, включая google, вероятно, будет довольно большим). Именно здесь вступает в действие REPARTITION_FACTOR. Идея состоит в том, что мы "запираем" каждое ядро, и поэтому, если человек заканчивается очень быстро, а один заканчивается медленно, у нас есть возможность разделить задачи между ними. В принципе, это хорошая идея.

Теперь давайте взглянем на размер одного раздела. Допустим, что все ваши данные имеют размер X МБ и у вас есть N разделов. Каждый раздел будет в среднем X/N MB. Если N велико относительно X, то у вас может быть очень небольшой средний размер раздела (например, несколько килобайт). В этом случае, как правило, рекомендуется снизить N, поскольку накладные расходы на управление каждым разделом становятся слишком высокими. С другой стороны, если размер очень большой (например, несколько ГБ), тогда вам необходимо хранить много данных одновременно, что может вызвать такие проблемы, как сбор мусора, высокая загрузка памяти и т.д.

Оптимальный размер - хороший вопрос, но, как правило, люди предпочитают разделы 100-1000 МБ, но, по правде говоря, десятки МБ, вероятно, также будут хороши.

Еще одна вещь, которую вы должны отметить, - это когда вы делаете расчет изменений ваших разделов. Например, скажем, вы начинаете с 1000 разделов по 100 Мбайт каждый, но затем фильтруете данные, чтобы каждый раздел становился 1K, тогда вы, вероятно, должны слиться. Подобные проблемы могут возникать, когда вы делаете группу или присоединяетесь. В таких случаях размер раздела и количество разделов изменяются и могут достигать нежелательного размера.