scala - Spark如何避免for循环？

Question

我是 Spark 新手，不明白 MapReduce 机制如何与 Spark 配合使用。我有一个只有 double 的 csv 文件，我想要的是用第一个向量与 rdd 的其余部分进行操作(计算欧几里得距离)。然后用其他向量进行迭代。除了这种方式之外还有其他方式吗？也许明智地使用笛卡尔积......

val rdd = sc.parallelize(Array((1,Vectors.dense(1,2)),(2,Vectors.dense(3,4),...)))
val array_vects = rdd.collect
val size = rdd.count
val emptyArray = Array((0,Vectors.dense(0))).tail
var rdd_rez = sc.parallelize(emptyArray)

for( ind <- 0 to size -1 ) {
   val vector = array_vects(ind)._2
   val rest = rdd.filter(x => x._1 != ind)
   val rdd_dist = rest.map( x => (x._1 , Vectors.sqdist(x._2,vector)))
   rdd_rez = rdd_rez ++ rdd_dist
}

感谢您的支持。

Answer 1

可以使用rdd.cartesian计算距离(所有向量对之间):

val rdd = sc.parallelize(Array((1,Vectors.dense(1,2)),
                               (2,Vectors.dense(3,4)),...))
val product = rdd.cartesian(rdd)

val result = product.filter{ case ((a, b), (c, d)) => a != c }
                    .map   { case ((a, b), (c, d)) => 
                                   (a, Vectors.sqdist(b, d)) }