在 PySpark 中按元素添加 RDD

Question

假设您有两个大小相同的向量，分别存储为 rdd1 和 rdd2。请编写一个函数，其中输入为 rdd1 和 rdd2，输出为 rdd，它是 rdd1 和 rdd2 的逐元素相加。您不应将所有数据加载到驱动程序。

提示：您可以在 Spark 中使用 zip()，而不是 Python 中的 zip()。

我不明白下面的代码有什么问题，是否正确。当我运行它时，它需要永远。你能帮我吗？谢谢

spark = SparkSession(sc)

numPartitions = 10
rdd1 = sc.textFile('./dataSet/points.txt',numPartitions).map(lambda x: int(x.split()[0]))
rdd2 = sc.textFile('./dataSet/points.txt',numPartitions).map(lambda x: int(x.split()[1]))


def ele_wise_add(rdd1, rdd2): 
    rdd3 = rdd1.zip(rdd2).map(lambda x,y: x + y)
    return rdd3

rdd3 = ele_wise_add(rdd1, rdd2)
print(rdd3.collect())

rdd1和rdd2各有10000个数，下面是其中的前10个数。

rdd1 = [47461, 93033, 92255, 33825, 90755, 3444, 48463, 37106, 5105, 68057]
rdd2 = [30614, 61104, 92322, 330, 94353, 26509, 36923, 64214, 69852, 63315]

expected output = [78075, 154137, 184577, 34155, 185108, 29953, 85386, 101320, 74957, 131372]

Answer 1

rdd1.zip(rdd2) 将为每对创建 一个元组 ，因此在编写 lambda 函数时，您只有 x 而没有 y .所以你想要 sum(x) 或 x[0] + x[1]，而不是 x + y。

rdd1 = spark.sparkContext.parallelize((47461, 93033, 92255, 33825, 90755, 3444, 48463, 37106, 5105, 68057))
rdd2 = spark.sparkContext.parallelize((30614, 61104, 92322, 330, 94353, 26509, 36923, 64214, 69852, 63315))

rdd1.zip(rdd2).map(lambda x: sum(x)).collect()
[78075, 154137, 184577, 34155, 185108, 29953, 85386, 101320, 74957, 131372]

在 PySpark 中按元素添加 RDD

Element-wise addition of RDDs in PySpark

apache-spark

dictionary

pyspark

lambda

zip