创建从使用火花的RDD

的元件的斯帕塞夫克托，我val rdd = RDD[(x: Int, y:Int), cov:Double]类型的Scala中，其中RDD的每个元素表示与x表示行，y表示列和cov表示矩阵的元素的数据结构元素的值：创建从使用火花的RDD

我需要从这个矩阵的行创建SparseVectors。所以我决定先RDD转换为RDD[x: Int, (y:Int, cov:Double)]，然后用groupByKey把特定行的所有元素结合在一起是这样的：

val rdd2 = rdd.map{case ((x,y),cov) => (x, (y, cov))}.groupByKey()

现在我需要创建SparseVectors：

val N = 7  //Vector Size 
val spvec = {(x: Int,y: Iterable[(Int, Double)]) => new SparseVector(N.toLong, Array(y.map(el => el._1.toInt)), Array(y.map(el => el._2.toDouble)))} 
val vecs = rdd2.map(spvec)

但是，这是弹出的错误。

type mismatch; found :Iterable[Int] required:Int 
type mismatch; found :Iterable[Double] required:Double

我猜测y.map(el => el._1.toInt)返回一个数组无法应用的迭代。我很感激，如果有人可以帮助如何做到这一点。

来源

2016-09-14 EdgeRover

最简单的解决方法是将转换为RowMatrix：

import org.apache.spark.mllib.linalg.distributed.{CoordinateMatrix, MatrixEntry} 

val rdd: RDD[((Int, Int), Double)] = ??? 

val vs: RDD[org.apache.spark.mllib.linalg.SparseVector]= new CoordinateMatrix(
    rdd.map{ 
    case ((x, y), cov) => MatrixEntry(x, y, cov) 
    } 
).toRowMatrix.rows.map(_.toSparse)

如果您想保留行标号可以使用toIndexedRowMatrix代替：

import org.apache.spark.mllib.linalg.distributed.IndexedRow 

new CoordinateMatrix(
    rdd.map{ 
    case ((x, y), cov) => MatrixEntry(x, y, cov) 
    } 
).toIndexedRowMatrix.rows.map { case IndexedRow(i, vs) => (i, vs.toSparse) }

来源

2016-09-14 20:05:46 zero323

谢谢。它适用于'toRowMatrix'，但不适用于'toIndexedRowMatrix'，表示'value toSparse不是org.apache.spark.mllib.linalg.distributed.IndexedRow'的成员。我确实想保留行索引。 – EdgeRover

因为它包含'IndexedRows'而不是'Vectors'。 – zero323

创建从使用火花的RDD

回答

相关问题