Перераспределение искры не работает, как ожидалось

Question

Я использую spark-sql 2.3.1, я установил

spark.sql.shuffle.partitions=40 

в моем коде '

val partitioned_df =  vals_df.repartition(col("model_id"),col("fiscal_year"),col("fiscal_quarter"))

Когда я говорю

println(" Number of partitions : " + partitioned_df.rdd.getNumPartitions)

Это дает 40 в качестве выхода, в действительности после перераспределения в идеале счет должен быть около 400. Почему перераспределение здесь не работает? Что я тут делаю не так? как это исправить?

Answer 1

set spark.sql.shuffle.partitions=40

Это относится только к JOINs и AGGregations, насколько я понимаю.

Попробуйте что-то вроде этого - мой собственный пример:

val df2 = df.repartition(40, $"c1", $"c2")

Вот вывод

val df2 = df.repartition(40, $"c1", $"c2").explain 

== Physical Plan ==
Exchange hashpartitioning(c1#114, c2#115, 40)
...

Может динамически устанавливать num разделов:

n = some calculation
val df2 = df.repartition(n, $"c1", $"c2").explain 

Answer 2

Этого и следовало ожидать, скаляр для repartition состояний:

Возвращает новый набор данных, разделенный на указанное разделение. выражений, используя spark.sql.shuffle.partitions в качестве числа перегородки. Полученный набор данных является хеш-секционированным.

Возможно, ваше понимание перераспределения неверно. При перераспределении по столбцу c все строки с одинаковым значением для c находятся в одном разделе, но один раздел может содержать несколько значений c