在两个不同的RDDS scala Spark中添加矢量

Question

我有两个带有此结构的RDD

org.apache.spark.rdd.RDD[(Long, org.apache.spark.mllib.linalg.Vector)]

在这里，RDD的每一行都包含索引Long和一个向量org.apache.spark.mllib.linalg.Vector。我想将Vector的每个组件添加到其他RDD中存在的其他Vector的相应组件中。第一个RDD的每个向量都应添加到其他RDD的每个向量。

的每个向量。

一个例子看起来像这样：

rdd1：

Array[(Long, org.apache.spark.mllib.linalg.Vector)] = 
      Array((0,[0.1,0.2]),(1,[0.3,0.4]))

rdd2：

Array[(Long, org.apache.spark.mllib.linalg.Vector)] = 
      Array((0,[0.3,0.8]),(1,[0.2,0.7]))

结果：

Array[(Long, org.apache.spark.mllib.linalg.Vector)] = 
Array((0,[0.4,1.0]),(0,[0.3,0.9]),(1,[0.6,1.2]),(1,[0.5,1.1]))

Answer 1

请使用列表而不是数组来考虑相同的情况。

这是我的解决方案：

    val l1 = List((0,List(0.1,0.2)),(1,List(0.1,0.2)))
    val l2 = List((0,List(0.3,0.8)),(1,List(0.2,0.7)))
    var sms = (l1 zip l2).map{ case (m, a) => (m._1, (m._2, a._2).zipped.map(_+_))}

让我们尝试数组：）

Answer 2

而不是驱动程序代码，您可以在转换中完成所有这些操作。如果您有大的RDD，这将有所帮助。这也会少做出弹药。

val a:RDD[(Long, org.apache.spark.mllib.linalg.Vector)]= sc.parallelize(Array((0l,Vectors.dense(0.1,0.2)),(1l,Vectors.dense(0.3,0.4))))
val b:RDD[(Long, org.apache.spark.mllib.linalg.Vector)]= sc.parallelize(Array((0l,Vectors.dense(0.3,0.8)),(1l,Vectors.dense(0.2,0.7))))
val ab= a join b
val result=ab.map(x => (x._1,Vectors.dense(x._2._1.apply(0)+x._2._2.apply(0),x._2._1.apply(1)+x._2._2.apply(1))))