在数据帧中使用 dropDuplicates 会导致分区号发生变化答案

【问题标题】：Using dropDuplicates in dataframe causes changes in the partition number在数据帧中使用 dropDuplicates 会导致分区号发生变化
【发布时间】：2016-09-24 02:49:27
【问题描述】：

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

df.rdd.getNumPartitions()
800

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

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

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

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

【问题讨论】：

Removing duplicates from rows based on specific columns in an RDD/Spark DataFrame的可能重复

标签： apache-spark pyspark apache-spark-sql partitioning

【解决方案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

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

【讨论】：

谢谢。我意识到我的错误是缺少 spark.sql.shuffle.partition 中的最后一个字符's'。

【解决方案2】：

我在Removing duplicates from rows based on specific columns in an RDD/Spark DataFrame找到了解决方案

使用 reduceByKey 而不是 dropDuplicates。 reduceByKey 还可以选择指定最终 rdd 的分区数。

在这种情况下使用 reduceByKey 的缺点是速度很慢。

【讨论】：