Скошенный набор данных присоединяется к Spark?

Ответ 1

Довольно хорошая статья о том, как это можно сделать: https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/

Краткая версия:

• Добавить случайный элемент в большой RDD и создать с ним новый ключ соединения
• Добавить случайный элемент в маленький RDD, используя explode/flatMap, чтобы увеличить количество записей и создать новый ключ соединения.
• Присоедините RDD к новому ключу присоединения, который теперь будет лучше распределен из-за случайного посева

Ответ 2

В зависимости от конкретного вида перекоса, который вы испытываете, могут быть разные способы его решения. Основная идея:

• Измените свой столбец соединения или создайте новый столбец объединения, который не перекошен, но который по-прежнему сохраняет адекватную информацию для соединения
• Соединяться в этом непересекающемся столбце - результирующие разделы не будут перекошены
• После объединения вы можете обновить столбец объединения до нужного формата или отбросить его, если вы создали новый столбец

Статья "Борьба с перекосом в искры", упомянутая в ответе LiMuBei, является хорошей техникой, если искаженные данные участвуют в объединении. В моем случае перекос был вызван очень большим количеством нулевых значений в столбце объединения. Нулевые значения не участвовали в объединении, но поскольку разделы Spark в столбце объединения, разделы после объединения были очень искажены, поскольку существовал один гигантский раздел, содержащий все нули.

Я решил это, добавив новый столбец, который изменил все нулевые значения на хорошо распределенное временное значение, такое как "NULL_VALUE_X", где X заменяется случайными числами от 1 до 10000, например. (на Java):

// Before the join, create a join column with well-distributed temporary values for null swids.  This column
// will be dropped after the join.  We need to do this so the post-join partitions will be well-distributed,
// and not have a giant partition with all null swids.
String swidWithDistributedNulls = "swid_with_distributed_nulls";
int numNullValues = 10000; // Just use a number that will always be bigger than number of partitions
Column swidWithDistributedNullsCol =
    when(csDataset.col(CS_COL_SWID).isNull(), functions.concat(
        functions.lit("NULL_SWID_"),
        functions.round(functions.rand().multiply(numNullValues)))
    )
    .otherwise(csDataset.col(CS_COL_SWID));
csDataset = csDataset.withColumn(swidWithDistributedNulls, swidWithDistributedNullsCol);

Затем присоединяется к этому новому столбцу, а затем после присоединения:

outputDataset.drop(swidWithDistributedNullsCol);

Ответ 3

Скажем, вы должны присоединиться к двум таблицам A и B на A.id = B.id. Предположим, что таблица A имеет перекос на id = 1.

т.е. выберите A.id из A join B на A.id = B.id

Существует два основных подхода к решению проблемы перекоса:

Подход 1:

Разбейте запрос/набор данных на две части - одну, содержащую только перекос, а другую, содержащую не перекошенные данные. В приведенном выше примере. запрос станет -

 1. select A.id from A join B on A.id = B.id where A.id <> 1;
 2. select A.id from A join B on A.id = B.id where A.id = 1 and B.id = 1;

Первый запрос не будет иметь никакого искажения, поэтому все задачи ResultStage будут завершены примерно в одно и то же время.

Если предположить, что B имеет всего несколько строк с B.id = 1, то он будет вписываться в память. Таким образом, второй запрос будет преобразован в широковещательное соединение. Это также называется присоединением к карте в Улье.

Ссылка: https://cwiki.apache.org/confluence/display/Hive/Skewed+Join+Optimization

Частичные результаты двух запросов затем могут быть объединены для получения окончательных результатов.

Подход 2:

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

1. Добавьте столбец в большую таблицу (A), скажем skewLeft и заполните его случайными числами от 0 до N-1 для всех строк.

2. Добавьте столбец в меньшую таблицу (B), скажем, skewRight. Повторите меньшую таблицу N раз. Таким образом, значения в новом столбце skewRight будут отличаться от 0 до N-1 для каждой копии исходных данных. Для этого вы можете использовать оператор explode sql/dataset.

После 1 и 2 присоединитесь к 2 наборам данных/таблицам с обновленным условием соединения to-

                *A.id = B.id && A.skewLeft = B.skewRight*

Ссылка: https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/

Ответ 4

Вы можете попытаться переделать "перекошенную" RDD на большее количество разделов или попытаться увеличить spark.sql.shuffle.partitions (по умолчанию 200).

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