如何计算 Apache Spark 中两个分布式 RowMatrix 的点积？答案

【问题标题】：How to compute the dot product of two distributed RowMatrix in Apache Spark?如何计算 Apache Spark 中两个分布式 RowMatrix 的点积？
【发布时间】：2017-09-04 13:48:06
【问题描述】：

让Q成为Spark中的分布式行矩阵，我想计算Q与其转置Q'。

然而，虽然行矩阵确实有一个 multiply() 方法，但它只能接受本地矩阵作为参数。

代码说明（Scala）：

val phi = new RowMatrix(phiRDD)            // phiRDD is an instance of RDD[Vector]
val phiTranspose = transposeRowMatrix(phi) // transposeRowMatrix()
                                           // returns the transpose of a RowMatrix
val crossMat = ?                           // phi * phiTranspose

请注意，我想执行 2 Distributed RowMatrix 的点积，而不是分布式行矩阵与本地行矩阵。

一种解决方案是使用IndexedRowMatrix，如下所示：

val phi = new IndexedRowMatrix(phiRDD)  // phiRDD is an instance of RDD[IndexedRow]
val phiTranspose = transposeMatrix(phi) // transposeMatrix()
                                        // returns the transpose of a Matrix
val crossMat = phi.toBlockMatrix().multiply( phiTranspose.toBlockMatrix()
                                             ).toIndexedRowMatrix()

但是，我想使用行矩阵方法，例如 tallSkinnyQR()，这意味着我应该使用.toRowMatrix() 方法将crossMat 转换为行矩阵：

val crossRowMat = crossMat.toRowMatrix()

终于可以申请了

crossRowMat.tallSkinnyQR()

但是这个过程包括分布式矩阵类型之间的许多转换，根据我从MLlib Programming Guide 了解到的，这是昂贵的：

选择正确的格式来存储大型分布式矩阵非常重要。将分布式矩阵转换为不同的格式可能需要全局 shuffle，这非常昂贵。

请有人详细说明一下。

【问题讨论】：

标签： apache-spark linear-algebra distributed-computing apache-spark-mllib matrix-multiplication

【解决方案1】：

只有支持矩阵-矩阵乘法的分布式矩阵是BlockMatrices。您必须相应地转换数据 - 人工索引就足够了：

new IndexedRowMatrix(
  rowMatrix.rows.zipWithIndex.map(x => IndexedRow(x._2,  x._1))
).toBlockMatrix match { case m => m.multiply(m.transpose) }

【讨论】：

【解决方案2】：

我使用了page 中列出的算法，它通过使用向量外积将乘法问题从点积问题转移到分布式标量积问题：

两个向量之间的外积是第二个向量与第一个向量中的所有元素，导致一个矩阵

我自己为行矩阵创建的乘法函数（可以更优化）就是这样结束的。

def multiplyRowMatrices(m1: RowMatrix, m2: RowMatrix)(implicit ctx: SparkSession): RowMatrix = {

 // Zip m1 columns with m2 rows
val m1Cm2R = transposeRowMatrix(m1).rows.zip(m2.rows)

// Apply scalar product between each entry in m1 vector with m2 row
val scalar = m1Cm2R.map{
case(column:DenseVector,row:DenseVector) => column.toArray.map{
  columnValue => row.toArray.map{
    rowValue => columnValue*rowValue
  }
 }
}

// Add all the resulting matrices point wisely
val sum = scalar.reduce{
case(matrix1,matrix2) => matrix1.zip(matrix2).map{
  case(array1,array2)=> array1.zip(array2).map{
    case(value1,value2)=> value1+value2
  }
 }
}

new RowMatrix(ctx.sparkContext.parallelize(sum.map(array=> Vectors.dense(array))))
}

之后我测试了这两种方法——我自己的函数和使用块矩阵——在一台机器上使用 300*10 矩阵

使用我自己的函数：

val PhiMat = new RowMatrix(phi)
val TphiMat = transposeRowMatrix(PhiMat)
val product = multiplyRowMatrices(PhiMat,TphiMat)

使用矩阵变换：

val MatRow = new RowMatrix(phi)
val MatBlock = new IndexedRowMatrix(MatRow.rows.zipWithIndex.map(x => IndexedRow(x._2,  x._1))).toBlockMatrix()
val TMatBlock = MatBlock.transpose
val productMatBlock = MatBlock.multiply(TMatBlock)
val productMatRow = productMatBlock.toIndexedRowMatrix().toRowMatrix()

第一种方法跨越 1 个工作，有 5 个阶段，总共用了 2s 完成。而第二种方法跨越 4 个作业，三个一个阶段和一个两个阶段，耗时 0.323 秒总共。第二种方法在随机读取/写入大小方面也优于第一种方法。

但我仍然对MLlib Programming指南声明感到困惑：

选择正确的格式来存储大容量和分布式矩阵。将分布式矩阵转换为不同的矩阵格式可能需要全局随机播放，这非常昂贵。

【讨论】：