Выполнение операции одного элемента над остальной частью RDD

Question

Я новичок в искры, и я действительно наслаждаюсь возможностями, предлагаемыми этой технологией. Моя проблема заключается в том, как выполнить операцию с одним элементом над остальной частью RDD для каждого элемента без использования цикла for. Вот моя попытка с циклом:

//RDD[Key:Int,Vector:(Double,Double)] 
val rdd = data.map(x => (x.split(',')(0).toInt,Vectors.dense(x.split(',')(1).toDouble,x.split(',')(2).toDouble))) 

for(ind <- 0 to rdd.count().toInt -1) { 
    val element1 = rdd.filter(x => x._1 == ind) 
    val vector1 = element1.first()._2 
    val rdd2 = rdd.map(x => { 
     var dist1 = Vectors.sqdist(x._2,vector1)  
     (x._1 , Math.sqrt(dist1)) 
     }) 
} 

Спасибо за ваши советы

Answer 1

Если вы ищете для поиска расстояния между всеми векторами, используйте rdd.cartesian:

import org.apache.spark.mllib.linalg.Vectors 

val rdd = sc.parallelize(Array("0,1.0,1.0","1,2.0,2.0","2,3.0,3.0")) 
val r = rdd.map(x => x.split(",")) 
      .map(y =>(y(0).toInt, Vectors.dense(y(1).toDouble, y(2).toDouble))) 

val res = r.cartesian(r).map{ case (first, second) => 
    ((first._1, second._1), 
    Math.sqrt(Vectors.sqdist(first._2, second._2))) 
} 

Однако он вычисляет, расстояния между одинаковыми векторами, дважды. (сначала (A, B), затем (B, A))