打印RDD/Dataframe的特定分区

Question

我一直在试验PySpark RDDs的分区和重新分区。

我注意到，当将一个小样本 RDD 从 2 个分区重新分区到 6 个分区时，只是添加了几个空部分。

rdd = sc.parallelize([1,2,3,43,54,678], 2)  
rdd.glom().collect()  
>>> [[1, 2, 3], [43, 54, 678]]

rdd6 = rdd.repartition(6)
rdd6.glom().collect()
>>> [[], [1, 2, 3], [], [], [], [43, 54, 678]]

现在，我想知道这是否也发生在我的真实数据中。

似乎我不能在较大的数据上使用glom（）（带有192497行的df）。
df.rdd.glom（）.collect（）
因为当我尝试时，什么也没发生。不过这是有道理的，由此产生的印刷品将是巨大的......

因此

我想打印每个分区，检查它们是否是空的。或者至少每个分区的前20个元素。

有什么想法吗？

PS：我找到了Spark的解决方案，但我无法让它们在PySpark中工作…
如何在Spark中打印特定RDD分区的元素？

顺便说一句：如果有人能向我解释为什么我会首先得到那些空的分区，我会全神贯注…
或者我如何知道什么时候会发生这种情况以及如何避免这种情况
或者，如果数据集中有空分区，它是否不会影响性能？

Answer 1

关于检查分区是否为空的可能性，我自己遇到了一些解决方案：

1. （如果没有那么多分区）

rdd.glom().collect()
>>>nothing happens

rdd.glom().collect()[1] 
>>>[1, 2, 3]

不过要小心，它会真正打印整个分区。对于我的数据，它导致了几千行打印。但它奏效了！

来源：如何在Spark中打印特定RDD分区的元素？

l = df.rdd.mapPartitionsWithIndex(lambda x,it: [(x,sum(1 for _ in it))]).collect()

min(l,key=lambda item:item[1])
>>>(2, 61705)

max(l,key=lambda item:item[1])
>>>(0, 65875)

来源：Spark Dataframes:连接后的倾斜分区

Answer 2

显然（令人惊讶），＜code＞rdd。重新分区仅执行合并，因此，没有混洗，难怪分布不均匀。一种方法是使用dataframe.repartition

rdd = sc.parallelize([1,2,3,43,54,678], 2)  
rdd.glom().collect()  
>>> [[1, 2, 3], [43, 54, 678]]

rdd6 = rdd.repartition(6)
rdd6.glom().collect()
>>> [[], [1, 2, 3], [], [], [], [43, 54, 678]]

rdd6_df = spark.createDataFrame(rdd, T.IntegerType()).repartition(6).rdd
rdd6_df.glom().collect()
[[Row(value=678)],
 [Row(value=3)],
 [Row(value=2)],
 [Row(value=1)],
 [Row(value=43)],
 [Row(value=54)]]