Как избежать цикла с искрой?

Question

Я новичок в искрах и не понимаю, как работает механизм трансляции с искру. У меня есть один файл 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 которые взять два плотных векторов и возвращает евклидовы расстояния.