如何计算火花中的距离矩阵？答案

【问题标题】：How to compute the distance matrix in spark?如何计算火花中的距离矩阵？
【发布时间】：2016-10-15 01:22:41
【问题描述】：

我尝试过配对样本，但它会消耗大量内存，因为 100 个样本会导致 9900 个样本，这更昂贵。在火花的分布式环境中计算距离矩阵的更有效方法是什么

这是我正在尝试的伪代码的 sn-p

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 个样本，通过配对 filterSamples（上面）会产生 4950 个样本，这对于大数据来说可能非常昂贵

【问题讨论】：

请发布您迄今为止尝试过的任何代码示例、与您的问题相关的数据或示例数据，以及您尝试过的任何库或资源。
我已经添加了代码的sn-p。希望它可以帮助您解决我的问题

标签： apache-spark distance-matrix bigdata

【解决方案1】：

我最近回答了一个类似的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 count = i.count
val indices = sc.parallelize(for(i <- 0L until count; j <- 0L until count; if i > j) yield (i, j))

val joined1 = indices.join(indexedData).map { case (i, (j, v)) => (j, (i,v)) }
val joined2 = joined1.join(indexedData).map { case (j, ((i,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) }

试试这个方法，并将它与您已经发布的方法进行比较。希望这可以加快您的代码速度。

【讨论】：

你拯救了我的一天！

【解决方案2】：

据我检查各种来源和Spark mllib clustering site 所见，Spark 目前不支持距离或 pdist 矩阵。

在我看来，100 个样本总是会输出至少 4950 个值；因此，使用转换（如 .map）手动创建分布式矩阵求解器将是最佳解决方案。

【讨论】：

【解决方案3】：

这可以作为jtitusj的answer的java版本..

public JavaPairRDD<Tuple2<Long, Long>, Double> getDistanceMatrix(Dataset<Row> ds, String vectorCol) {

    JavaRDD<Vector> rdd = ds.toJavaRDD().map(new Function<Row, Vector>() {

        private static final long serialVersionUID = 1L;

        public Vector call(Row row) throws Exception {
            return row.getAs(vectorCol);
        }

    });

    List<Vector> vectors = rdd.collect();

    long count = ds.count();

    List<Tuple2<Tuple2<Long, Long>, Double>> distanceList = new ArrayList<Tuple2<Tuple2<Long, Long>, Double>>();

    for(long i=0; i < count; i++) {
        for(long j=0; j < count && i > j; j++) {
            Tuple2<Long, Long> indexPair = new Tuple2<Long, Long>(i, j);
            double d = DistanceMeasure.getDistance(vectors.get((int)i), vectors.get((int)j));
            distanceList.add(new Tuple2<Tuple2<Long, Long>, Double>(indexPair, d));
        }
    }

    return distanceList;
}

【讨论】：