为什么Spark会因java.lang.OutOfMemoryError而失败:超出GC开销限制？

Question

我正在尝试实现一个在Spark之前工作正常的Hadoop Map/Reduce作业.Spark应用程序定义如下:

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( _ + _)

哪里MyFunctions.combine是

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)

combine如果用于输入的列表很大,则该函数会生成大量映射键,这就是抛出异常的位置.

在Hadoop Map Reduce设置中,我没有遇到任何问题,因为这是combine函数产生的点,Hadoop将映射对写入磁盘.Spark似乎会将所有内容保留在内存中,直到它爆炸为止java.lang.OutOfMemoryError: GC overhead limit exceeded.

我可能正在做一些非常基本的错误,但我找不到任何关于如何从这个方面挺身而出的指示,我想知道如何避免这种情况.由于我是Scala和Spark的总菜鸟,我不确定问题是来自一个还是来自另一个,或两者兼而有之.我目前正在尝试在我自己的笔记本电脑上运行这个程序,它适用于tuples数组长度不长的输入.提前致谢.

Answer 1

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

-Dspark.executor.memory=6g

您还可以考虑在创建实例时明确设置工作器数SparkContext:

分布式集群

在以下位置设置从属名称conf/slaves:

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

Answer 2

正如已经建议的那样,调整内存可能是一个很好的方法,因为这是一种昂贵的操作,可以以丑陋的方式扩展.但也许一些代码更改将有所帮助.

您可以在组合函数中采用不同的方法,if通过使用该combinations函数来避免语句.在组合操作之前,我还将元组的第二个元素转换为双精度:

tuples.

    // Convert to doubles only once
    map{ x=>
        (x._1, x._2.toDouble)
    }.

    // Take all pairwise combinations. Though this function
    // will not give self-pairs, which it looks like you might need
    combinations(2).

    // Your operation
    map{ x=>
        (toKey(x{0}._1, x{1}._1), x{0}._2*x{1}._2)
    }

这将提供一个迭代器,您可以使用下游,或者,如果您愿意,可以使用转换为列表(或其他内容)toList.

Answer 3

在文档(http://spark.apache.org/docs/latest/running-on-yarn.html)中,您可以阅读如何配置执行程序和内存限制.例如:

--master yarn-cluster --num-executors 10 --executor-cores 3 --executor-memory 4g --driver-memory 5g  --conf spark.yarn.executor.memoryOverhead=409

memoryOverhead应该是执行程序内存的10%.

编辑:修正了4096到409(以下评论指的是这个)