我有一组点（每个点都是文本文件中的一行），它们分布在 Spark 的分区之间。
我使用 mapPartitions 操作，它在每个分区中得到一半的点（并不重要为什么和如何）。
现在，我想成对合并分区，因此分区 #1 和 #2 将是合并分区，分区 #3 和 #4 将是第二个合并分区，依此类推。
我将继续运行 mapPartitions，直到我只留下几个分区。 如何使用 Spark 做到这一点？
这里与 Hadoop 的类似物是我成对合并输出文件并再次运行任务。
我会再次澄清一下：我有 x 个分区，在我运行 mapPartitions 之后， 我想成对合并它们，所以我会有 x/2 个分区并再次运行 mapPartitions 等等..
【问题讨论】：
标签： apache-spark
使用treeAggregate 作为模型应该可以解决问题：
from math import log
def binaryReduce(rdd, f):
assert log(rdd.getNumPartitions(), 2) % 1 == 0
def mapPartition(i, iter):
i = i / 2
for x in iter:
yield i, x
while rdd.getNumPartitions() != 1:
rdd = (rdd
.mapPartitionsWithIndex(mapPartition)
.reduceByKey(f, rdd.getNumPartitions() / 2)
.values())
return rdd.first()
如果您更喜欢更明确的方法，您可以一直保留分区号：
def binaryReduce(rdd, f):
assert log(rdd.getNumPartitions(), 2) % 1 == 0
def initPartition(i, iter):
for x in iter:
yield i, x
rdd = rdd.mapPartitionsWithIndex(initPartition)
while rdd.getNumPartitions() != 1:
rdd = (rdd
.reduceByKey(f)
.map(lambda x: (x[0] / 2, x[1]))
.partitionBy(rdd.getNumPartitions() / 2))
return rdd.values().reduce(f)
对于 Python 3.0+，请务必将 / 替换为 //。
【讨论】：
what is the difference between both approaches you showed me here - 你可以说第一个取决于实现的细节，否则没有实际区别。
这个问题很模糊。 如果我做对了，您可以在将数据映射到 (key,value) 后尝试 reduceByKey。
(http://spark.apache.org/docs/latest/programming-guide.html#parallelized-collections)
希望这会有所帮助。
编辑：您应该将 mapPartitionsWithIndex(func) 与必须为 (Int, Iterator) => Iterator 类型的 func 一起使用。
【讨论】：