Spark 重新分区未按预期工作

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

这是

意料之中的，repartition状态的scaladoc：

返回按给定分区分区的新数据集 表达式，使用 Spark.sql.shuffle.partitions 作为 分区。生成的数据集是哈希分区的。

也许您对重新分区的理解是错误的。当您按列 c 重新分区时，具有相同值的所有行c都位于同一分区中，但 1 个分区可以保存多个值c

Answer 2

set spark.sql.shuffle.partitions=40

这仅适用于 JOIN 和聚合，这是我的理解。

尝试这样的事情 - 我自己的例子：

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

这是输出

val df2 = df.repartition(40, $"c1", $"c2").explain 
== Physical Plan ==
Exchange hashpartitioning(c1#114, c2#115, 40)
...

可以动态设置分区数：

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