apache-spark - 在pyspark中使用缓冲区收集RDD

Question

我想要一种从我的 RDD 一次(或小批量)返回行的方法，以便我可以根据需要在本地收集行。我的 RDD 足够大，无法放入名称节点的内存中，因此运行 collect()会导致错误。

有没有办法重新创建 collect()操作但使用生成器，以便将来自 RDD 的行传递到缓冲区？另一种选择是 take()一次来自缓存的 RDD 的 100000 行，但我不认为 take()允许你指定一个开始位置？

Answer 1

最好的选择是使用 RDD.toLocalIterator 当时只收集一个分区。它创建了一个标准的 Python 生成器:

rdd = sc.parallelize(range(100000))
iterator = rdd.toLocalIterator()
type(iterator)

## generator

even = (x for x in iterator if not x % 2)

您可以使用特定的分区器并调整多个分区来调整在单个批次中收集的数据量。

不幸的是，它是有代价的。要收集小批量，您必须启动多个 Spark 作业，而且非常昂贵。所以一般来说，当时收集元素不是一种选择。