如何避免使用 Spark 的 for 循环？

Question

我是 spark 新手，不了解 mapreduce 机制如何与 spark 一起工作。我有一个只有双打的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)) }

Answer 2

我不认为你为什么要尝试做那样的事情。您可以简单地按照以下方式执行此操作。

val initialArray = Array( ( 1,Vectors.dense(1,2) ), ( 2,Vectors.dense(3,4) ),... )

val firstVector = initialArray( 0 )

val initialRdd = sc.parallelize( initialArray )

val euclideanRdd = initialRdd.map( { case ( i, vec ) => ( i, euclidean( firstVector, vec ) ) } )

我们定义了一个函数euclidean，它接受两个密集向量并返回欧几里得距离。