spark中 进行高维矩阵的SVD分解（2）

由于大型矩阵的奇异值分解，在普通的单机上是无法计算的，所以只能在spark集群上计算

在集群上计算会遇到一个问题，调用 “ M.computeSVD(5000, true,1.0E-9d)” 时，A=U*s*V 分解的s向量存储，V矩阵存储正常，但是分解的左奇异矩阵U默认为rowmatrix矩阵，这个矩阵存储的（存储代码：U.rows.saveAsTextFile("hdfs://10.67.89.240:9000/outsvd/big_UUT1")）时候，发现存储结果为为好几部分 eg：part-00000、

part-00001、part-00002.....等，所以想办法让输出结果为一个 （存储代码：U.rows.repartition(1).saveAsTextFile("hdfs://10.67.89.240:9000/outsvd/big_UUT1")）

但是发现我求出的矩阵U 里面应用的顺序不对，比如按道理是 part-00001+part-00000+part-00002 的顺序 ，但是结果按照part-00000+part-00001+part-00002的顺序来的，

尝试着输出多次发现U矩阵的结果是不确定的 就是因为 几个输出的part部分汇总的问题，所以一直不知道该怎么搞，还是希望懂的人能够给与建议

我这里是一个替代方案：就是先计算 A矩阵的svd分解存储 分解的右奇异矩阵V，在对A进行转秩，再次存储V，这样就相当于很好的存储了A矩阵的 U 、V奇异矩阵了，但是效率不高了

//************************************************************************************************************************************************

//************************************************************************************************************************************************

//先对原始数据求转秩，再计算矩阵的svd分解，保存计算的结果保存 v矩阵，即没转秩时的u矩阵

//************************************************************************************************************************************************

//************************************************************************************************************************************************

import java.util.{Date, Locale}

import java.text.DateFormat

import java.text.DateFormat._

import org.apache.spark.mllib.linalg.{Vector, Vectors}

import org.apache.spark.mllib.linalg.Vector

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

import org.apache.spark.mllib.linalg._

val nowT1 = new Date

val dm = sc.textFile("hdfs:///usr/matrix/title_matrix.txt").map(_.split(',').map(_.toDouble)).map(line => Vectors.dense(line))

def transpose(m:Array[Array[Double]]):Array[Array[Double]]=

{

(for{c<-m(0).indices}

yield{m.map(_(c))}).toArray}

val ma=dm.map(_.toArray).take(dm.count.toInt)

val tma=transpose(ma)

//tma= tma.toArray.grouped(tma.numRows).toList.transpose

sc.makeRDD(tma, 1).zipWithIndex().map(line => line._1.mkString(",")).saveAsTextFile("hdfs:///usr/matrix/bigt")

val M = new RowMatrix(sc.textFile("hdfs:///usr/matrix/bigt/part-00000").map(_.split(','))

.map(_.map(_.toDouble))

.map(_.toArray).map(line => Vectors.dense(line)))

val svd = M.computeSVD(5000, true,1.0E-9d)

val lastT1 = new Date

val U = svd.U

U.rows.repartition(1).saveAsTextFile("hdfs://10.67.89.240:9000/outsvd/big_UUT1")

sc.makeRDD(svd.s.toArray, 1).saveAsTextFile("hdfs://10.67.89.240:9000/outsvd/big_SST1")

val V = svd.V.toArray.grouped(svd.V.numRows).toList.transpose

sc.makeRDD(V, 1).zipWithIndex().map(line => line._1.mkString(",")).saveAsTextFile("hdfs://10.67.89.240:9000/outsvd/big_VVT1")

//************************************************************************************************************************************************

//对原始数据，计算矩阵的svd计分解，保存计算的结果保存 v矩阵，s向量 则存在的结果

//************************************************************************************************************************************************

val now1 = new Date

val M = new RowMatrix(sc.textFile("hdfs:///usr/matrix/title_matrix.txt").map(_.split(','))

.map(_.map(_.toDouble))

.map(_.toArray).map(line => Vectors.dense(line)))

val svd = M.computeSVD(1000, true,1.0E-9d)

val last1 = new Date

val U = svd.U

U.rows.repartition(1).saveAsTextFile("hdfs://10.67.89.240:9000/outsvd/big_UU1")

sc.makeRDD(svd.s.toArray, 1).saveAsTextFile("hdfs://10.67.89.240:9000/outsvd/big_SS1")

val V = svd.V.toArray.grouped(svd.V.numRows).toList.transpose

sc.makeRDD(V, 1).zipWithIndex().map(line =>line._1.mkString(",")).saveAsTextFile("hdfs://10.67.89.240:9000/outsvd/big_VV1")

//nowT1

//lastT1

now1

last1

//************************************************************************************************************************************************

//************************************************************************************************************************************************