スパークを使用RDD

Question

の要素からスパーセベクターを作成し、IはRDDの各要素はyを表す列とcovを表し、行を表すxと行列の要素を表すスカラ、タイプval rdd = RDD[(x: Int, y:Int), cov:Double]のデータ構造を有しています要素の値：

この行列の行からスパースベクトルを作成する必要があります。だから私は、最初のRDD[x: Int, (y:Int, cov:Double)]にRDDを変換し、このように一緒に特定の行のすべての要素を入れて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)が配列に適用することはできません反復可能を返していることを推測しています。私は誰かがこれを行う方法を手伝ってくれたらと感謝します。

Answer 1

最も簡単な解決策は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) }