无法访问 pyspark 中的流水线 Rdd [重复]答案

【问题标题】：Can not access Pipelined Rdd in pyspark [duplicate]无法访问 pyspark 中的流水线 Rdd [重复]
【发布时间】：2017-12-09 22:18:21
【问题描述】：

我正在尝试使用 pyspark 从头开始实现 K-means。我正在对 rdd 执行各种操作，但是当我尝试显示最终处理的 rdd 的结果时，出现一些错误，如“管道式 RDD 无法迭代”或类似的错误，并且 .collect() 之类的东西不再起作用，因为管道化的 rdd 问题。

from __future__ import print_function
import sys
import numpy as np
def closestPoint(p, centers):
    bestIndex = 0
    closest = float("+inf")
    for i in range(len(centers)):
        tempDist = np.sum((p - centers[i]) ** 2)
        if tempDist < closest:
            closest = tempDist
            bestIndex = i
    return bestIndex

data=SC.parallelize([1, 2, 3,5,7,3,5,7,3,6,4,66,33,66,22,55,77])

K = 3
convergeDist = float(0.1)

kPoints = data.takeSample(False, K, 1)
tempDist = 1.0

while tempDist > convergeDist:
    closest = data.map(
        lambda p: (closestPoint(p, kPoints), (p, 1)))



    pointStats = closest.reduceByKey(
        lambda p1_c1, p2_c2: (p1_c1[0] + p2_c2[0], p1_c1[1] + p2_c2[1]))

    newPoints = pointStats.map(
        lambda st: (st[0], st[1][0] / st[1][1]))
    print(newPoints)


    tempDist = sum(np.sum((kPoints[iK] - p) ** 2) for (iK, p) in newPoints).collect()

       # tempDist = sum(np.sum((kPoints[iK] - p) ** 2) for (iK, p) in newPoints)




    for (iK, p) in newPoints:
        kPoints[iK] = p

print("Final centers: " + str(kPoints))

我得到的错误是：

TypeError: 'PipelinedRDD' 对象不可迭代

【问题讨论】：

您正在混合使用 Spark 和 Numpy。只选一个
kPoints.collect() 可以正常工作。我不明白你关于这是一个问题的观点
这一行似乎是问题所在： tempDist = sum(np.sum((kPoints[iK] - p) ** 2) for (iK, p) in newPoints).collect() 。 ..显然你不能使用for循环/迭代newPoints（这是一个流水线RDD）
通常在常规 RDD 上工作的东西，比如简单的对方付费电话似乎不适用于 newPoints 管道 RDD。

标签： numpy apache-spark pyspark rdd

【解决方案1】：

您不能迭代 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

这不起作用，因为 test 是一个 RDD。另一方面，如果您通过操作将数据带回驱动程序，现在它将成为您可以迭代的对象，例如：

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

你去吧，调用一个动作并将数据带回驱动程序，小心不要有太多数据，否则你可能会出现内存不足的异常

【讨论】：