apache-spark - 在数据帧中使用 dropDuplicates 会导致分区号发生变化

Question

我有一个包含 800 个分区的大型数据框。

df.rdd.getNumPartitions()
800

当我在数据帧上使用 dropDuplicates 时，它会将分区更改为默认 200

df = df.dropDuplicates()
df.rdd.getNumPartitions()
200

这种行为给我带来了问题，因为它会导致内存不足。

你对解决这个问题有什么建议吗？我尝试将 spark.sql.shuffle.partition 设置为 800 但它不起作用。谢谢

Answer 1

这是因为 dropDuplicates需要洗牌。如果你想获得特定数量的分区，你应该设置 spark.sql.shuffle.partitions (其默认值为 200)

df = sc.parallelize([("a", 1)]).toDF()
df.rdd.getNumPartitions()
## 8

df.dropDuplicates().rdd.getNumPartitions()
## 200

sqlContext.setConf("spark.sql.shuffle.partitions", "800")

df.dropDuplicates().rdd.getNumPartitions()
## 800

另一种方法(Spark 1.6+)是先重新分区:

df.repartition(801, *df.columns).dropDuplicates().rdd.getNumPartitions()
## 801

由于不执行本地聚合，因此稍微灵活但效率较低。