規模矩陣/星火

Question

我有以下列表

id1, column_index1, value1 
id2, column_index2, value2 
... 

我轉變爲一個索引行矩陣執行以下操作：

val data_mapped = data.map({ case (id, col, score) => (id, (col, score))}) 
val data_mapped_grouped = data_mapped.groupByKey 
val indexed_rows = data_mapped_grouped.map({ case (id, vals) => IndexedRow(id, Vectors.sparse(nCols.value, vals.toSeq))}) 
val mat = new IndexedRowMatrix(indexed_rows) 

我想執行這個矩陣一些預處理：從每列中刪除列的總和，通過其差異來標準化每列。 我曾嘗試使用內置的標準定標器

val scaler = new StandardScaler().fit(indexed_rows.map(x => x.features)) 

，但這似乎並沒有

感謝您的幫助，有可能與IndexedRow型！

Answer 1

據我從你的問題的理解，這裏就是你需要做的，在你的IndexedRow

import org.apache.spark.mllib.feature.{StandardScaler, StandardScalerModel} 
import org.apache.spark.mllib.linalg.distributed.IndexedRow 
import org.apache.spark.mllib.linalg.{Vector, Vectors} 
import org.apache.spark.rdd.RDD 

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

object nCol { 
    val value: Int = ??? 
} 

val data_mapped: RDD[(Int, (Int, Double))] = 
    data.map({ case (id, col, score) => (id, (col, score)) }) 
val data_mapped_grouped: RDD[(Int, Iterable[(Int, Double)])] = 
    data_mapped.groupByKey 

val indexed_rows: RDD[IndexedRow] = data_mapped_grouped.map { 
     case (id, vals) => 
     IndexedRow(id, Vectors.sparse(nCol.value, vals.toSeq)) 
} 

執行StandardScaler適合你可以從你的IndexedRow用一個簡單的地圖讓你的載體是什麼

val vectors: RDD[Vector] = indexed_rows.map { case i: IndexedRow => i.vector } 

既然您已經擁有了一個RDD [矢量]，您可以嘗試使用它的縮放器。

val scaler: StandardScalerModel = new StandardScaler().fit(vectors) 

我希望這有助於！