如何在Apache Spark中计算RowMatrix的逆函数

Question

我有一个X，分布式矩阵，采用RowMatrix形式。我正在使用Spark 1.3.0。我需要能够计算 X 逆。

Answer 1

import org.apache.spark.mllib.linalg.{Vectors,Vector,Matrix,SingularValueDecomposition,DenseMatrix,DenseVector}
import org.apache.spark.mllib.linalg.distributed.RowMatrix
def computeInverse(X: RowMatrix): DenseMatrix = {
  val nCoef = X.numCols.toInt
  val svd = X.computeSVD(nCoef, computeU = true)
  if (svd.s.size < nCoef) {
    sys.error(s"RowMatrix.computeInverse called on singular matrix.")
  }
  // Create the inv diagonal matrix from S 
  val invS = DenseMatrix.diag(new DenseVector(svd.s.toArray.map(x => math.pow(x,-1))))
  // U cannot be a RowMatrix
  val U = new DenseMatrix(svd.U.numRows().toInt,svd.U.numCols().toInt,svd.U.rows.collect.flatMap(x => x.toArray))
  // If you could make V distributed, then this may be better. However its alreadly local...so maybe this is fine.
  val V = svd.V
  // inv(X) = V*inv(S)*transpose(U)  --- the U is already transposed.
  (V.multiply(invS)).multiply(U)
  }

Answer 2

我在使用带有选项的函数时遇到问题

conf.set("spark.sql.shuffle.partitions", "12")

RowMatrix 中的行被打乱了。

这是一个对我有用的更新

import org.apache.spark.mllib.linalg.{DenseMatrix,DenseVector}
import org.apache.spark.mllib.linalg.distributed.IndexedRowMatrix
def computeInverse(X: IndexedRowMatrix)
: DenseMatrix = 
{
  val nCoef = X.numCols.toInt
  val svd = X.computeSVD(nCoef, computeU = true)
  if (svd.s.size < nCoef) {
    sys.error(s"IndexedRowMatrix.computeInverse called on singular matrix.")
  }
  // Create the inv diagonal matrix from S 
  val invS = DenseMatrix.diag(new DenseVector(svd.s.toArray.map(x => math.pow(x, -1))))
  // U cannot be a RowMatrix
  val U = svd.U.toBlockMatrix().toLocalMatrix().multiply(DenseMatrix.eye(svd.U.numRows().toInt)).transpose
  val V = svd.V
  (V.multiply(invS)).multiply(U)
}

Answer 3

X.computeSVD返回的矩阵 U 的维度为 m x k，其中 m 是原始（分布式）RowMatrix X 的行数。人们会期望 m 很大（可能大于 k），因此如果我们希望我们的代码扩展到非常大的 m 值，则不建议在驱动程序中收集它。

我会说下面的两个解决方案都存在这个缺陷。@ 给出的答案Alexander Kharlamov调用val U = svd.U.toBlockMatrix().toLocalMatrix()，它收集驱动程序中的矩阵。@ Climbs_lika_Spyder给出的答案也会发生同样的情况（顺便说一句，你的昵称摇滚！！），它调用svd.U.rows.collect.flatMap(x => x.toArray).我宁愿建议依赖分布式矩阵乘法，例如此处发布的 Scala 代码。