如何从Spark RDD创建共生矩阵

Question

我想从一些元组创建一个共生矩阵,见下文:

val rdd = sc.parallelize(Array(
  Array("101","103","105"), Array("102","105"),Array("101","102","103","105"))
val coocMatrix = new ArrayBuffer[(String, Int)]()

// map
rdd.collect().foreach(x => 
  { for(i <- 0 to x.length-1)
  { for(j <- i+1 to x.length-1)
  { coocMatrix += (x(i)+"#"+x(j), 1) }}

// convert to rdd again
val rdd2 = sc.parallelize(coocMatrix)

// reduce
vall matrix = rdd2.collect().groupByKey()

所以我们最终获得了以下数据

(101#103,2),(101#105,2),(102#105,2),(101#102,1),
(103#105,2),(102#103,1),(102#105,1)

这个算法非常慢,因为当有200万个元组时,它是O(n*n)并且是不可行的.有没有算法来计算这种共生矩阵？

Answer 1

该combinations方法允许您提取给定数组中出现的对列表.之后你可以减少ByKey

    rdd.flatMap{_.combinations(2).map{pairs=>(pairs.mkString("#"),1)}}.
    reduceByKey(_+_)