pyspark：'PipelinedRDD' 对象不可迭代答案

【问题标题】：pyspark: 'PipelinedRDD' object is not iterablepyspark：'PipelinedRDD' 对象不可迭代
【发布时间】：2016-08-11 00:55:32
【问题描述】：

我收到此错误，但我不知道为什么。基本上我从这段代码中出错了：

    a = data.mapPartitions(helper(locations))

其中 data 是一个 RDD，我的助手定义为：

    def helper(iterator, locations): 
        for x in iterator:
            c = locations[x]
            yield c

（位置只是一个数据点数组）我看不出问题出在哪里，但我也不是 pyspark 方面的佼佼者，所以有人可以告诉我为什么我得到的“PipelinedRDD”对象不能从此代码中迭代吗？

【问题讨论】：

您不能以您的方式迭代 rdd。请看stackoverflow.com/questions/25914789/…
@Mohan ：谢谢，我想我现在明白了，但我仍然遇到同样的错误。我现在称之为：a = data.mapPartitions(lambda iterator: helper(iterator, locations))。我还做错了什么？

标签： pyspark rdd

【解决方案1】：

RDD 可以使用 map 和 lambda 函数进行迭代。我已经使用以下方法迭代了流水线 RDD

lines1 = sc.textFile("\..\file1.csv")
lines2 = sc.textFile("\..\file2.csv")

pairs1 = lines1.map(lambda s: (int(s), 'file1'))
pairs2 = lines2.map(lambda s: (int(s), 'file2'))

pair_result = pairs1.union(pairs2)

pair_result.reduceByKey(lambda a, b: a + ','+ b)

result = pair.map(lambda l: tuple(l[:1]) + tuple(l[1].split(',')))
result_ll = [list(elem) for elem in result]

===> result_ll = [list(elem) for elem in result]

TypeError: 'PipelinedRDD' 对象不可迭代

我使用 map 函数替换了迭代

result_ll = result.map( lambda elem: list(elem))

希望这有助于相应地修改您的代码

【讨论】：

【解决方案2】：

我更喜欢下面链接的另一个问题中所说的答案： Can not access Pipelined Rdd in pyspark

您不能迭代 RDD，您需要首先调用一个操作来将您的数据返回给驱动程序。快速示例：

`>>> test = sc.parallelize([1,2,3])
 >>> for i in test:
     ...    print i
     ... 
     Traceback (most recent call last):
     File "<stdin>", line 1, in <module>
     TypeError: 'RDD' object is not iterable`

但是例如你可以使用'.collect()'

`>>> for i in test.collect():
     ...      print i
 1                                                                               
 2
 3`

【讨论】：