Pyspark 数据帧重新分区将所有数据放在一个分区中答案

【问题标题】：Pyspark dataframe repartitioning puts all data in one partitionPyspark 数据帧重新分区将所有数据放在一个分区中
【发布时间】：2018-01-08 07:01:14
【问题描述】：

我有一个架构如下的数据框：

root
 |-- category_id: string (nullable = true)
 |-- article_title: string (nullable = true)

看起来像这样的数据：

+-----------+--------------------+
|category_id|     articletitle   |
+-----------+--------------------+
|       1000|HP EliteOne 800 G...|
|       1000|ASUS  EB1501P ATM...|
|       1000|HP EliteOne 800 G...|
|          1|ASUS R557LA-XO119...|
|          1|HP EliteOne 800 G...|
+-----------+--------------------+

只有两个不同的category_id 1000 和 1。

我想通过category_id 和mapPartition 对每个分区进行重新分区。

p_df = df.repartition(2, "category_id")
p_df.rdd.mapPartitionsWithIndex(some_func)

但是数据没有得到正确的分区，预期的结果是每个 mappartition 将只有一个category_id 的数据。但实际结果是一个分区得到0条记录，而另一个分区得到所有记录。

为什么会发生这种情况以及如何解决这个问题？

已经有一个question 介绍了 spark 分区器的工作原理。我的问题不同，因为答案仅包含对分区器如何工作的解释，但我的问题是关于为什么会发生这种情况（已经回答）以及如何解决它。

【问题讨论】：

您是如何得出一个分区为空而另一个分区有所有记录的结论的？可以添加p_df.withColumn("partition" , spark_partition_id()).show()的输出吗？
没关系。它为 Spark 1.6 提供了准确的分区，但为 Spark 2.2 中的所有记录提供了相同的分区 ID。

标签： apache-spark pyspark

【解决方案1】：

您正确使用了repartition 和mapPartitionsWithIndex 函数。

如果你将explain函数应用为

df.repartition(2, "category_id").explain()

您将看到以下输出，清楚地表明它已重新分区为两个分区。

== Physical Plan ==
Exchange hashpartitioning(category_id#0L, 2)
+- Scan ExistingRDD[category_id#0L,articletitle#1L]

现在真正的罪魁祸首是 hashPartitioning，它将 1、10、1000、100000 ... 视为与分区号相同的哈希 2

解决方案是将分区数更改为 3 或更多，

或

将category_id 1000 更改为其他值。

【讨论】：

【解决方案2】：

@Ramesh Maharjan 在上述答案中解释了重新分区将所有数据放在一个分区中的原因。更多关于哈希分区here

我能够通过使用自定义分区器将数据转到不同的分区器。我将rdd变成pairRdd格式（category_id，row）并使用partitionBy方法给出分区数和custom_partitioner。

    categories = input_df.select("category_id").distinct().rdd.map(lambda r: r.category_id).collect()
    cat_idx = dict([(cat, idx) for idx, cat in enumerate(categories)])

    def category_partitioner(cid):
        return cat_idx[cid]

【讨论】：