Skip to content

[SPARK-12869] Implemented an improved version of the toIndexedRowMatrix #10839

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Fokko wants to merge 6 commits into from
Closed

[SPARK-12869] Implemented an improved version of the toIndexedRowMatrix #10839

Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4d7c297
Implemented an improved version of the toIndexedRowMatrix method of t…
Jan 19, 2016
fe1842e
Updated the method based on the suggestion of @mengxr
Mar 13, 2016
c043e77
Resolved conflicts
Mar 17, 2016
d3c780d
Updated the code based to make use of the Matrix iterator
Mar 17, 2016
6c0b58d
Processed the feedback of @mengxr and added tests for the mapping of …
Fokko Mar 21, 2016
25c5f66
Merged master
Fokko Mar 21, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 24 additions & 4 deletions mllib/src/main/scala/org/apache/spark/mllib/linalg/distributed/BlockMatrix.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,11 @@ package org.apache.spark.mllib.linalg.distributed

import scala.collection.mutable.ArrayBuffer

import breeze.linalg.{DenseMatrix => BDM, Matrix => BM}
import breeze.linalg.{DenseMatrix => BDM, DenseVector => BDV, Matrix => BM, SparseVector => BSV, Vector => BV}

import org.apache.spark.{Logging, Partitioner, SparkException}
import org.apache.spark.annotation.Since
import org.apache.spark.mllib.linalg.{DenseMatrix, Matrices, Matrix, SparseMatrix}
import org.apache.spark.mllib.linalg._
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel

Expand Down Expand Up @@ -263,13 +263,33 @@ class BlockMatrix @Since("1.3.0") (
new CoordinateMatrix(entryRDD, numRows(), numCols())
}


/** Converts to IndexedRowMatrix. The number of columns must be within the integer range. */
@Since("1.3.0")
def toIndexedRowMatrix(): IndexedRowMatrix = {
require(numCols() < Int.MaxValue, "The number of columns must be within the integer range. " +
s"numCols: ${numCols()}")
// TODO: This implementation may be optimized
toCoordinateMatrix().toIndexedRowMatrix()

val rows = blocks.flatMap { case ((blockRowIdx, blockColIdx), mat) =>
mat.rowIter.zipWithIndex.map {
case (vector, rowIdx) =>
blockRowIdx * rowsPerBlock + rowIdx -> (blockColIdx, vector.toBreeze)
}
}.groupByKey().map { case (rowIdx, vectors) =>

val wholeVector = vectors.head match {

@mengxr mengxr Mar 21, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The output vector type should depend on the total number of active elements (or nonzeros) instead of the first one. Could you try vectors.map(_.activeSize).sum and compare it with numCols to decide which vector type to use?

case (idx, v: BDV[_]) => BDV.zeros[Double](numCols().toInt)

@mengxr mengxr Mar 21, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

numCols() are now called several times in this method. We should save the value to a local variable at the beginning of the method.

case (idx, v: BSV[_]) => BSV.zeros[Double](numCols().toInt)
case _ => throw new SparkException(s"Cannot convert an empty vector to an indexed row")
}

vectors.foreach { case (blockColIdx: Int, vec: BV[Double]) =>
val offset = colsPerBlock * blockColIdx
wholeVector(offset until offset + vec.size) := vec

@mengxr mengxr Mar 21, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(minor) vec.size should be the same as colsPerBlock.

}
new IndexedRow(rowIdx, Vectors.fromBreeze(wholeVector))
}
new IndexedRowMatrix(rows)
}

/** Collect the distributed matrix on the driver as a `DenseMatrix`. */
Expand Down