少量列和行上的 Spark PCA OutOfMemory 错误

Question

我正在尝试在具有 2168 列和大量行的 RowMatrix 上执行 Spark MLLib PCA（使用 Scala）。但是，我观察到即使矩阵中只有 2 行（一个 112KB 的文本文件），在同一个工作步骤中总是会产生以下错误：

Exception in thread "main" java.lang.OutOfMemoryError: Java heap space 
        at breeze.linalg.svd$.breeze$linalg$svd$$doSVD_Double(svd.scala:92) 
        at breeze.linalg.svd$Svd_DM_Impl$.apply(svd.scala:39) 
        at breeze.linalg.svd$Svd_DM_Impl$.apply(svd.scala:38) 
        at breeze.generic.UFunc$class.apply(UFunc.scala:48) 
        at breeze.linalg.svd$.apply(svd.scala:22) 
        at org.apache.spark.mllib.linalg.distributed.RowMatrix.computePrincipalComponents(RowMatrix.scala:380) 
        at SimpleApp$.main(scala-pca.scala:17) 
        at SimpleApp.main(scala-pca.scala) 
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
        at java.lang.reflect.Method.invoke(Method.java:601) 
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:569) 
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:166) 
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:189) 
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:110) 
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

我还观察到，无论 RowMatrix 中的行数如何，使用 1100 列或更少的列可以解决此错误。

我在 21 个节点上独立运行 Spark 1.3.0，每个节点有 12 个工作器和 20G 内存。我通过spark-submit 和--driver-memory 6g 和--conf spark.executor.memory=1700m 提交工作。在spark-env.sh 中设置了以下选项：

SPARK_WORKER_MEMORY=1700M
SPARK_WORKER_CORES=1
SPARK_WORKER_INSTANCES=12

这是我提交的代码：

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.mllib.linalg.Matrix
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.mllib.linalg.{Vector, Vectors}

object SimpleApp {
  def main(args: Array[String]) {
    val datafilePattern = "/path/to/data/files*.txt"
    val conf = new SparkConf().setAppName("pca_analysis").setMaster("master-host")
    val sc = new SparkContext(conf)
    val lData = sc.textFile(datafilePattern).cache()

    val vecData = lData.map(line => line.split(" ").map(v => v.toDouble)).map(arr => Vectors.dense(arr))
    val rmat: RowMatrix = new RowMatrix(vecData)
    val pc: Matrix = rmat.computePrincipalComponents(15)
    val projected: RowMatrix = rmat.multiply(pc)

    println("Finished projecting rows.")
  }
}

有没有其他人在使用 computePrincipalComponents() 方法时遇到过这个问题？非常感谢任何帮助。

Answer 1

我刚刚遇到了这个问题，解决这个问题的方法是在需要时将--driver-memory 增加到 2G 或更多。