在 PySpark 中按元素添加 RDD

Element-wise addition of RDDs in PySpark

假设您有两个大小相同的向量,分别存储为 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]

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]