Spark：查找 RDD 的每个分区大小

Posted 2023-04-15

【中文标题】Spark：查找 RDD 的每个分区大小

【英文标题】：Spark: Find Each Partition Size for RDD

【发布时间】：2016-12-09 20:11:35

【问题描述】：

找到给定 RDD 的每个分区大小的最佳方法是什么。我正在尝试调试一个倾斜的分区问题，我试过这个：

l = builder.rdd.glom().map(len).collect()  # get length of each partition
print('Min Parition Size: ',min(l),'. Max Parition Size: ', max(l),'. Avg Parition Size: ', sum(l)/len(l),'. Total Partitions: ', len(l))

它适用于小型 RDD，但对于较大的 RDD，它会给出 OOM 错误。我的想法是glom() 导致这种情况发生。但无论如何，只是想知道是否有更好的方法来做到这一点？

【参考方案1】：

用途：

builder.rdd.mapPartitions(lambda it: [sum(1 for _ in it)])

【讨论】：

如果您能解释一下 OP 的方法在他的情况下不起作用，以及为什么这种方法 起作用...？

它给出错误“AttributeError：'PipelinedRDD'对象没有属性'_get_object_id'”

【参考方案2】：

虽然@LostInOverflow 的回答效果很好。我找到了另一种方法来查找每个分区的大小和索引，使用下面的代码。感谢this awesome post.

代码如下：

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

然后您可以使用以下代码获取最大和最小大小的分区：

min(l,key=lambda item:item[1])
max(l,key=lambda item:item[1])

找到偏斜分区的键，如果需要，我们可以进一步调试该分区的内容。