Как вычислить матрицу расстояния в искрах?

Question

Я пробовал спаривать образцы, но это стоит огромного объема памяти, поскольку 100 образцов приводят к 9900 образцам, что является более дорогостоящим. Что может быть более эффективным способом вычисления матрицы расстояний в распределенной среде в искре

Вот фрагмент псевдокода, что я пытаюсь

val input = (sc.textFile("AirPassengers.csv",(numPartitions/2))) 
val i = input.map(s => (Vectors.dense(s.split(',').map(_.toDouble)))) 
val indexed = i.zipWithIndex()                  //Including the index of each sample 
val indexedData = indexed.map{case (k,v) => (v,k)} 

val pairedSamples = indexedData.cartesian(indexedData) 

val filteredSamples = pairedSamples.filter{ case (x,y) => 
(x._1.toInt > y._1.toInt) //to consider only the upper or lower trainagle 
} 
filteredSamples.cache 
filteredSamples.count 

Приведенный выше код создает пар, но даже если мой набор данных содержит 100 образцов, путем спаривания фильтрованных образцов (см. Выше), приводит к выборке 4950, что может быть очень дорогостоящим для больших данных.

Answer 1

Насколько я могу судить по проверке различных источников и Spark mllib clustering site, Spark в настоящее время не поддерживает матрицы расстояния или pdist ,

На мой взгляд, 100 образцов всегда будут выводить не менее 4950 значений; поэтому оптимальным решением было бы вручную создание распределенного матричного решателя с использованием преобразования (например .map).

Answer 2

Я недавно ответил на аналогичный question.

В принципе, он придет к вычислениям n(n-1)/2 пар, которые будут в вашем примере вычислять 4950. Однако, чем отличается этот подход, я использую объединения вместо cartesian. С вашего кода, решение будет выглядеть следующим образом:

val input = (sc.textFile("AirPassengers.csv",(numPartitions/2))) 
val i = input.map(s => (Vectors.dense(s.split(',').map(_.toDouble)))) 
val indexed = i.zipWithIndex() 

//Including the index of each sample 
val indexedData = indexed.map { case (k,v) => (v,k) } 

//prepare indices 
val cnt = i.count 
val indices = sc.parallelize((for(i <- 0 until cnt; j <- 0 until cnt; if (i > j)))) 

val joined1 = indices.join(indexedData).map { case (i, (j, v)) => (j, (i,v)) } 
val joined2 = joined1.join(indexedData).map { case (j, ((j,v1),v2)) => ((i,j),(v1,v2)) } 

// after that, you can then compute the distance using your distFunc 
val distRDD = joined2.mapValues{ case (v1, v2) => distFunc(v1, v2) } 

Попробуйте этот метод и сравнить его с тем, кого уже размещен. Надеюсь, это немного ускорит ваш код.