From 39b0b227d3bedfc789743972f5a0ded8583e8e7b Mon Sep 17 00:00:00 2001 From: Yuhao Yang Date: Tue, 5 May 2015 19:51:51 +0800 Subject: [PATCH 1/4] initial draft for discussion --- .../mllib/linalg/distributed/RowMatrix.scala | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala b/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala index 9a89a6f3a515f..c120216d65b37 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala @@ -497,6 +497,65 @@ class RowMatrix( columnSimilaritiesDIMSUM(computeColumnSummaryStatistics().normL2.toArray, gamma) } + /** + * Compute QR decomposition for rowMatrix. The implementation is designed to optimize the QR + * decomposition (factorizations) for the RowMatrix of a tall and skinny shape, yet it applies + * to RowMatrix in general. + * + * Reference: + * Austin R. Benson, David F. Gleich, James Demmel. "Direct QR factorizations for tall-and + * -skinny matrices in MapReduce architectures", 2013 IEEE International Conference on Big Data + * @param computeQ: whether to computeQ, which is quite expensive. + * @return the decomposition result as (Option[Q], R), where Q is a RowMatrix and R is Matrix. + */ + def TSQR(computeQ: Boolean = false): (Option[RowMatrix], Matrix) = { + val col = numCols().toInt + + // split rows horizontally into smaller matrices, and compute QR for each of them + val blockQRs = rows.mapPartitions(rowsIterator =>{ + val partRows = rowsIterator.toArray + val rowCount = partRows.size + var bdm = BDM.zeros[Double](partRows.size, col) + var i = 0 + partRows.foreach(row =>{ + bdm(i, ::) := row.toBreeze.t + i += 1 + }) + + val blockQR = breeze.linalg.qr.reduced(bdm) + Iterator((blockQR.r, blockQR.q)) + }).cache + + // combine the R part from previous results horizontally into a tall matrix + val blockRsRdd = blockQRs.map(_._1).collect() + val CombinedR = blockRsRdd.reduceLeft((r1, r2) => BDM.vertcat(r1, r2)) + + val CombinedRDecomposition = breeze.linalg.qr.reduced(CombinedR) + val finalR = Matrices.fromBreeze(CombinedRDecomposition.r.toDenseMatrix) + + val finalQ = if(computeQ){ + val blockQ = blockQRs.map(_._2) + val rightPartQ = CombinedRDecomposition.q + val rightQArray = (0 until blockQ.count().toInt) + .map(i => rightPartQ(i * col until (i + 1) * col, ::)) + .toArray + val rightQrdd = blockQ.context.parallelize(rightQArray) + + val qProducts = blockQ.zip(rightQrdd).map(m => m._1 * m._2) + val newRows = qProducts.flatMap(m => { + val row = m.rows + (0 until row).map(i =>{ + val bv = m(i, ::).t + Vectors.fromBreeze(bv) + }) + }) + Some(new RowMatrix(newRows)) + } + else None + + (finalQ, finalR) + } + /** * Find all similar columns using the DIMSUM sampling algorithm, described in two papers * From 3fbdb61fbd0881fb5036294c55291066211987fd Mon Sep 17 00:00:00 2001 From: Yuhao Yang Date: Wed, 29 Jul 2015 17:58:54 +0800 Subject: [PATCH 2/4] update qr to indirect and add ut --- .../linalg/SingularValueDecomposition.scala | 8 ++ .../mllib/linalg/distributed/RowMatrix.scala | 83 ++++++++----------- .../linalg/distributed/RowMatrixSuite.scala | 13 +++ 3 files changed, 55 insertions(+), 49 deletions(-) diff --git a/mllib/src/main/scala/org/apache/spark/mllib/linalg/SingularValueDecomposition.scala b/mllib/src/main/scala/org/apache/spark/mllib/linalg/SingularValueDecomposition.scala index 9669c364bad8f..b416d50a5631e 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/linalg/SingularValueDecomposition.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/linalg/SingularValueDecomposition.scala @@ -25,3 +25,11 @@ import org.apache.spark.annotation.Experimental */ @Experimental case class SingularValueDecomposition[UType, VType](U: UType, s: Vector, V: VType) + +/** + * :: Experimental :: + * Represents QR factors. + */ +@Experimental +case class QRDecomposition[UType, VType](Q: UType, R: VType) + diff --git a/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala b/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala index a0ab25995e673..3a7c1af87a865 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala @@ -22,7 +22,7 @@ import java.util.Arrays import scala.collection.mutable.ListBuffer import breeze.linalg.{DenseMatrix => BDM, DenseVector => BDV, SparseVector => BSV, axpy => brzAxpy, - svd => brzSvd} + svd => brzSvd, MatrixSingularException, inv} import breeze.numerics.{sqrt => brzSqrt} import com.github.fommil.netlib.BLAS.{getInstance => blas} @@ -498,62 +498,47 @@ class RowMatrix( } /** - * Compute QR decomposition for rowMatrix. The implementation is designed to optimize the QR - * decomposition (factorizations) for the RowMatrix of a tall and skinny shape, yet it applies - * to RowMatrix in general. - * + * Compute QR decomposition for [[RowMatrix]]. The implementation is designed to optimize the QR + * decomposition (factorization) for the [[RowMatrix]] of a tall and skinny shape. * Reference: - * Austin R. Benson, David F. Gleich, James Demmel. "Direct QR factorizations for tall-and - * -skinny matrices in MapReduce architectures", 2013 IEEE International Conference on Big Data - * @param computeQ: whether to computeQ, which is quite expensive. - * @return the decomposition result as (Option[Q], R), where Q is a RowMatrix and R is Matrix. + * Paul G. Constantine, David F. Gleich. "Tall and skinny QR factorizations in MapReduce + * architectures" ([[http://dx.doi.org/10.1145/1996092.1996103]]) + * + * @param computeQ: whether to computeQ + * @return QRDecomposition(Q, R), Q = null if computeQ = false. */ - def TSQR(computeQ: Boolean = false): (Option[RowMatrix], Matrix) = { + def tallSkinnyQR(computeQ: Boolean = false): QRDecomposition[RowMatrix, Matrix] = { val col = numCols().toInt - // split rows horizontally into smaller matrices, and compute QR for each of them - val blockQRs = rows.mapPartitions(rowsIterator =>{ - val partRows = rowsIterator.toArray - val rowCount = partRows.size - var bdm = BDM.zeros[Double](partRows.size, col) + val blockQRs = rows.glom().map{ partRows => + val bdm = BDM.zeros[Double](partRows.length, col) var i = 0 - partRows.foreach(row =>{ + partRows.foreach{ row => bdm(i, ::) := row.toBreeze.t i += 1 - }) - - val blockQR = breeze.linalg.qr.reduced(bdm) - Iterator((blockQR.r, blockQR.q)) - }).cache - - // combine the R part from previous results horizontally into a tall matrix - val blockRsRdd = blockQRs.map(_._1).collect() - val CombinedR = blockRsRdd.reduceLeft((r1, r2) => BDM.vertcat(r1, r2)) - - val CombinedRDecomposition = breeze.linalg.qr.reduced(CombinedR) - val finalR = Matrices.fromBreeze(CombinedRDecomposition.r.toDenseMatrix) - - val finalQ = if(computeQ){ - val blockQ = blockQRs.map(_._2) - val rightPartQ = CombinedRDecomposition.q - val rightQArray = (0 until blockQ.count().toInt) - .map(i => rightPartQ(i * col until (i + 1) * col, ::)) - .toArray - val rightQrdd = blockQ.context.parallelize(rightQArray) - - val qProducts = blockQ.zip(rightQrdd).map(m => m._1 * m._2) - val newRows = qProducts.flatMap(m => { - val row = m.rows - (0 until row).map(i =>{ - val bv = m(i, ::).t - Vectors.fromBreeze(bv) - }) - }) - Some(new RowMatrix(newRows)) + } + breeze.linalg.qr.reduced(bdm).r + }.cache() + + // combine the R part from previous results vertically into a tall matrix + val combinedR = blockQRs.treeReduce((r1, r2) => BDM.vertcat(r1, r2)) + val breezeR = breeze.linalg.qr.reduced(combinedR).r.toDenseMatrix + val finalR = Matrices.fromBreeze(breezeR) + val finalQ = if (computeQ) { + try { + val invR = inv(breezeR) + this.multiply(Matrices.fromBreeze(invR)) + } + catch { + case err: MatrixSingularException => + logWarning("R is not invertible and return Q as null") + null + } } - else None - - (finalQ, finalR) + else { + null + } + QRDecomposition(finalQ, finalR) } /** diff --git a/mllib/src/test/scala/org/apache/spark/mllib/linalg/distributed/RowMatrixSuite.scala b/mllib/src/test/scala/org/apache/spark/mllib/linalg/distributed/RowMatrixSuite.scala index b6cb53d0c743e..804aa0b15e620 100644 --- a/mllib/src/test/scala/org/apache/spark/mllib/linalg/distributed/RowMatrixSuite.scala +++ b/mllib/src/test/scala/org/apache/spark/mllib/linalg/distributed/RowMatrixSuite.scala @@ -19,6 +19,7 @@ package org.apache.spark.mllib.linalg.distributed import scala.util.Random +import breeze.numerics.abs import breeze.linalg.{DenseVector => BDV, DenseMatrix => BDM, norm => brzNorm, svd => brzSvd} import org.apache.spark.SparkFunSuite @@ -238,6 +239,18 @@ class RowMatrixSuite extends SparkFunSuite with MLlibTestSparkContext { } } } + + test("QR Decomposition") { + for (mat <- Seq(denseMat, sparseMat)) { + val result = mat.tallSkinnyQR(true) + val expected = breeze.linalg.qr.reduced(mat.toBreeze()) + val calcQ = result.Q + val calcR = result.R + assert(closeToZero(abs(expected.q) - abs(calcQ.toBreeze()))) + assert(closeToZero(abs(expected.r) - abs(calcR.toBreeze.asInstanceOf[BDM[Double]]))) + assert(closeToZero(calcQ.multiply(calcR).toBreeze - mat.toBreeze())) + } + } } class RowMatrixClusterSuite extends SparkFunSuite with LocalClusterSparkContext { From 0fb1012859dda102742c94f2f1894e5baf75a392 Mon Sep 17 00:00:00 2001 From: Yuhao Yang Date: Thu, 30 Jul 2015 08:50:43 +0800 Subject: [PATCH 3/4] hierarchy R computing --- .../mllib/linalg/distributed/RowMatrix.scala | 17 +++++++++-------- .../linalg/distributed/RowMatrixSuite.scala | 4 ++++ 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala b/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala index 3a7c1af87a865..a4259fc29c698 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala @@ -510,32 +510,33 @@ class RowMatrix( def tallSkinnyQR(computeQ: Boolean = false): QRDecomposition[RowMatrix, Matrix] = { val col = numCols().toInt // split rows horizontally into smaller matrices, and compute QR for each of them - val blockQRs = rows.glom().map{ partRows => + val blockQRs = rows.glom().map { partRows => val bdm = BDM.zeros[Double](partRows.length, col) var i = 0 - partRows.foreach{ row => + partRows.foreach { row => bdm(i, ::) := row.toBreeze.t i += 1 } breeze.linalg.qr.reduced(bdm).r - }.cache() + } // combine the R part from previous results vertically into a tall matrix - val combinedR = blockQRs.treeReduce((r1, r2) => BDM.vertcat(r1, r2)) + val combinedR = blockQRs.treeReduce{ (r1, r2) => + val partialR = BDM.vertcat(r1, r2) + breeze.linalg.qr.reduced(partialR).r + } val breezeR = breeze.linalg.qr.reduced(combinedR).r.toDenseMatrix val finalR = Matrices.fromBreeze(breezeR) val finalQ = if (computeQ) { try { val invR = inv(breezeR) this.multiply(Matrices.fromBreeze(invR)) - } - catch { + } catch { case err: MatrixSingularException => logWarning("R is not invertible and return Q as null") null } - } - else { + } else { null } QRDecomposition(finalQ, finalR) diff --git a/mllib/src/test/scala/org/apache/spark/mllib/linalg/distributed/RowMatrixSuite.scala b/mllib/src/test/scala/org/apache/spark/mllib/linalg/distributed/RowMatrixSuite.scala index 804aa0b15e620..283ffec1d49d7 100644 --- a/mllib/src/test/scala/org/apache/spark/mllib/linalg/distributed/RowMatrixSuite.scala +++ b/mllib/src/test/scala/org/apache/spark/mllib/linalg/distributed/RowMatrixSuite.scala @@ -249,6 +249,10 @@ class RowMatrixSuite extends SparkFunSuite with MLlibTestSparkContext { assert(closeToZero(abs(expected.q) - abs(calcQ.toBreeze()))) assert(closeToZero(abs(expected.r) - abs(calcR.toBreeze.asInstanceOf[BDM[Double]]))) assert(closeToZero(calcQ.multiply(calcR).toBreeze - mat.toBreeze())) + // Decomposition without computing Q + val rOnly = mat.tallSkinnyQR(computeQ = false) + assert(rOnly.Q == null) + assert(closeToZero(abs(expected.r) - abs(rOnly.R.toBreeze.asInstanceOf[BDM[Double]]))) } } } From cec797b4d1d6af444001593a04e32e769ef422ec Mon Sep 17 00:00:00 2001 From: Yuhao Yang Date: Thu, 30 Jul 2015 09:59:09 +0800 Subject: [PATCH 4/4] remove unnecessary qr --- .../spark/mllib/linalg/distributed/RowMatrix.scala | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala b/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala index a4259fc29c698..bfc90c9ef8527 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/RowMatrix.scala @@ -504,7 +504,7 @@ class RowMatrix( * Paul G. Constantine, David F. Gleich. "Tall and skinny QR factorizations in MapReduce * architectures" ([[http://dx.doi.org/10.1145/1996092.1996103]]) * - * @param computeQ: whether to computeQ + * @param computeQ whether to computeQ * @return QRDecomposition(Q, R), Q = null if computeQ = false. */ def tallSkinnyQR(computeQ: Boolean = false): QRDecomposition[RowMatrix, Matrix] = { @@ -522,14 +522,13 @@ class RowMatrix( // combine the R part from previous results vertically into a tall matrix val combinedR = blockQRs.treeReduce{ (r1, r2) => - val partialR = BDM.vertcat(r1, r2) - breeze.linalg.qr.reduced(partialR).r + val stackedR = BDM.vertcat(r1, r2) + breeze.linalg.qr.reduced(stackedR).r } - val breezeR = breeze.linalg.qr.reduced(combinedR).r.toDenseMatrix - val finalR = Matrices.fromBreeze(breezeR) + val finalR = Matrices.fromBreeze(combinedR.toDenseMatrix) val finalQ = if (computeQ) { try { - val invR = inv(breezeR) + val invR = inv(combinedR) this.multiply(Matrices.fromBreeze(invR)) } catch { case err: MatrixSingularException =>