如何在 pyspark 中使用 mapPartitions

Question

遵循 Apache Spark 文档后，我尝试使用 mapPartition 模块。在下面的代码中，我希望看到函数 myfunc 中的初始 RDD，我只是在打印值后返回迭代器。但是当我在 RDD 上执行 collect 时，它是空的。

from pyspark import SparkConf
from pyspark import SparkContext                                                          

def myfunc(it):
    print(it.next())
    return it

def fun1(sc):
    n = 5
    rdd = sc.parallelize([x for x in range(n+1)], n)
    print(rdd.mapPartitions(myfunc).collect())


if __name__ == "__main__":                                                                
    conf = SparkConf().setMaster("local[*]")                                              
    conf = conf.setAppName("TEST2")                                                       
    sc = SparkContext(conf = conf)                                                        
    fun1(sc)

Answer 1

mapPartitions 在这里不相关。迭代器（这里是itertools.chain）是有状态的，只能被遍历一次。当您调用 it.next() 时，您会读取并丢弃第一个元素，返回的是序列的尾部。

当分区只有一项时（应该是除一项之外的所有项目），您实际上丢弃了整个分区。

一些注意事项：

• 将任何内容放入任务中的标准输出通常是没有用的。
• 您使用 next 的方式不可移植，不能在 Python 3 中使用。