Почему я получаю так много пустых разделов при повторном копировании Spark Dataframe? - PullRequest
Новая
       37

Почему я получаю так много пустых разделов при повторном копировании Spark Dataframe?

0 голосов
/ 05 июня 2018

Я хочу разбить фрейм данных "df1" на 3 столбца.Этот фрейм данных имеет ровно 990 уникальных комбинаций для этих 3 столбцов: 

In [17]: df1.createOrReplaceTempView("df1_view")

In [18]: spark.sql("select count(*) from (select distinct(col1,col2,col3) from df1_view) as t").show()
+--------+                                                                      
|count(1)|
+--------+
|     990|
+--------+

Чтобы оптимизировать обработку этого фрейма данных, я хочу разделить df1, чтобы получить 990 секций, по одному на каждую ключевую возможность: 

In [19]: df1.rdd.getNumPartitions()
Out[19]: 24

In [20]: df2 = df1.repartition(990, "col1", "col2", "col3")

In [21]: df2.rdd.getNumPartitions()
Out[21]: 990

Я написал простой способ подсчета строк в каждом разделе: 

In [22]: def f(iterator):
    ...:     a = 0
    ...:     for partition in iterator:
    ...:         a = a + 1
    ...:     print(a)
    ...: 

In [23]: df2.foreachPartition(f)

И я заметил, что на самом деле я получил 628 разделов с одним или несколькими значениями ключа и 362пустые разделы.

Я предполагал, что spark будет перераспределять равномерно (1 значение ключа = 1 раздел), но это не похоже на это, и я чувствую, что это перераспределение добавляет перекос данных, хотя это должно бытьнаоборот ...

Какой алгоритм использует Spark для разделения кадра данных на столбцы?Есть ли способ достичь того, что я считал возможным?

Я использую Spark 2.2.0 на Cloudera.

1 Ответ

0 голосов
/ 05 июня 2018

Чтобы распределить данные по разделам спарк, нужно каким-то образом преобразовать значение столбца в индекс раздела.В Spark есть два разделителя по умолчанию - HashPartitioner и RangePartitioner.Различные преобразования в Spark могут применять разные разделители - например, join будет применять разделитель хеш-функции.

В основном для формулы разделителя хеш-функции преобразование значения в индекс раздела будет value.hashCode() % numOfPartitions.В вашем случае несколько значений отображаются на один и тот же индекс раздела.

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

...