为什么Spark会在java.lang.OutofMemoryError：超过GC开销限制时失败？

Question

val data = spark.textFile(file, 2).cache()
val result = data
  .map(//some pre-processing)
  .map(docWeightPar => (docWeightPar(0),docWeightPar(1))))
  .flatMap(line => MyFunctions.combine(line))
  .reduceByKey( _ + _)

def combine(tuples: Array[(String, String)]): IndexedSeq[(String,Double)] =
  for (i <- 0 to tuples.length - 2;
       j <- 1 to tuples.length - 1
  ) yield (toKey(tuples(i)._1,tuples(j)._1),tuples(i)._2.toDouble * tuples(j)._2.toDouble)

我可能做了一些非常基本的错误，但我找不到任何关于如何从这一点上站出来的指针，我想知道我如何才能避免这一点。由于我是Scala和Spark的新手，我不确定问题是来自其中一个还是另一个，或者两者都有。我目前正在我自己的笔记本电脑上运行这个程序，它适用于tuples数组长度不是很长的输入。提前谢了。

Answer 1

在启动spark-shell或spark-submit时添加以下JVM arg：

-Dspark.executor.memory=6g

您还可以考虑在创建SparkContext实例时显式设置worker的数量：

在conf/slaves中设置从属名称：

val sc = new SparkContext("master", "MyApp")