大矩陣運算：Scala中的乘法/ Apache Spark

Question

我需要乘以兩個大矩陣，X和Y。典型地，X具有〜500K行和〜18K列，並且Y具有〜18K行和〜18K列。預計矩陣X是稀疏的，並且矩陣Y預計是稀疏/密集的。在Scala/Apache Spark中執行這種乘法的理想方式是什麼？

Answer 1

我有一些代碼給你。它將一個矩陣表示爲一個列向量數組（這意味着數組中的每個條目都是一列，而不是一行）。將兩個1000 * 1000矩陣乘以大約0.7s。兩個10,000 * 10,000矩陣需要11分鐘。 20,000 * 20,000小時1.5小時，（500k * 18k）次（18k * 18k）30小時。但如果你並行運行它（通過使用註釋掉的代碼），它應該運行速度提高2到3倍（在4核心CPU上）。但請記住，第一個矩陣中的列數必須與第二個矩陣中的行數相同。

class Matrix(val columnVectors: Array[Array[Double]]) { 
    val columns = columnVectors.size 
    val rows = columnVectors.head.size 
    def *(v: Array[Double]): Array[Double] = { 
    val newValues = Array.ofDim[Double](rows) 
    var col = 0 
    while(col < columns) { 
     val n = v(col) 
     val column = columnVectors(col) 
     var row = 0 
     while(row < newValues.size) { 
     newValues(row) += column(row) * n 
     row += 1 
     } 
     col += 1 
    } 
    newValues 
    } 
    def *(other: Matrix): Matrix = { 
    //do the calculation on only one cpu 
    new Matrix(other.columnVectors.map(col => this * col)) 

    //do the calculation in parallel on all available cpus 
    //new Matrix(other.columnVectors.par.map(col => this * col).toArray) 
    } 
    override def toString = { 
    columnVectors.transpose.map(_.mkString(", ")).mkString("\n") 
    } 
} 

編輯：

好吧，這裏是一個更好的版本。我現在將行向量存儲在矩陣中而不是列向量中。這使得在第一個矩陣稀疏的情況下優化乘法更容易。 此外我使用迭代器添加了一個惰性版本的矩陣乘法。由於第一個矩陣是500k * 18k = 90億個數字，所以這樣一個懶惰的版本將允許您在不需要太多內存的情況下進行乘法運算。您只需創建一個可以懶惰地讀取行的迭代器，例如從數據庫開始，然後將結果迭代器中的行寫回。

import scala.collection.Iterator 
import scala.util.{Random => rand} 

def time[T](descr: String)(f: => T): T = { 
    val start = System.nanoTime 
    val r = f 
    val end = System.nanoTime 
    val time = (end - start)/1e6 
    println(descr + ": time = " + time + "ms") 
    r 
} 

object Matrix { 
    def mulLazy(m1: Iterator[Array[Double]], m2: Matrix): Iterator[Array[Double]] = { 
    m1.grouped(8).map { group => 
     group.par.map(m2.mulRow).toIterator 
    }.flatten 
    } 
} 

class Matrix(val rowVectors: Array[Array[Double]]) { 
    val columns = rowVectors.head.size 
    val rows = rowVectors.size 

    private def mulRow(otherRow: Array[Double]): Array[Double] = { 
    val rowVectors = this.rowVectors 
    val result = Array.ofDim[Double](columns) 
    var i = 0 
    while(i < otherRow.size) { 
     val value = otherRow(i) 
     if(value != 0) { //optimization for sparse matrix 
     val row = rowVectors(i) 
     var col = 0 
     while(col < result.size) { 
      result(col) += value * row(col) 
      col += 1 
     } 
     } 
     i += 1 
    } 
    result 
    } 

    def *(other: Matrix): Matrix = { 
    new Matrix(rowVectors.par.map(other.mulRow).toArray) 
    } 

    def equals(other: Matrix): Boolean = { 
    java.util.Arrays.deepEquals(this.rowVectors.asInstanceOf[Array[Object]], other.rowVectors.asInstanceOf[Array[Object]]) 
    } 

    override def equals(other: Any): Boolean = { 
    if(other.isInstanceOf[Matrix]) equals(other.asInstanceOf[Matrix]) else false 
    } 

    override def toString = { 
    rowVectors.map(_.mkString(", ")).mkString("\n") 
    } 
} 

def randMatrix(rows: Int, columns: Int): Matrix = { 
    new Matrix((1 to rows).map(_ => Array.fill(columns)(rand.nextDouble * 100)).toArray) 
} 

def sparseRandMatrix(rows: Int, columns: Int, ratio: Double): Matrix = { 
    new Matrix((1 to rows).map(_ => Array.fill(columns)(if(rand.nextDouble > ratio) 0 else rand.nextDouble * 100)).toArray) 
} 

val N = 2000 

val m1 = sparseRandMatrix(N, N, 0.1) // only 10% of the numbers will be different from 0 
val m2 = randMatrix(N, N) 

val m3 = m1.rowVectors.toIterator 

val m12 = time("m1 * m2")(m1 * m2) 
val m32 = time("m3 * m2")(Matrix.mulLazy(m3, m2)) //doesn't take much time because the matrix multiplication is lazy 

println(m32) 

println("m12 == m32 = " + (new Matrix(m32.toArray) == m12))