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

【问题标题】：Why does Spark fail with java.lang.OutOfMemoryError: GC overhead limit exceeded?为什么 Spark 会因 java.lang.OutOfMemoryError 失败：超出 GC 开销限制？
【发布时间】：2015-02-12 06:01:59
【问题描述】：

我正在尝试实现之前在 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 数组的长度不是很长的输入。提前致谢。

【问题讨论】：

“数据”已经是 RDD 了吗？
我刚刚编辑了代码以显示我如何加载数据。
好的，我继续回答，因为已经假设它实际上是（如您所示）一个 RDD
很高兴看到您的帖子末尾包含最终解决方案，谢谢！
最终的解决方案是我用接受的答案标记的那个:)

标签： scala apache-spark

【解决方案1】：

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

-Dspark.executor.memory=6g

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

分布式集群

在conf/slaves中设置slave名字：

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

【讨论】：

不幸的是，内存选项没有帮助。我仍然遇到内存不足异常。
在我添加此选项并将最小分区更改为 100 spark.textFile(conceptsFile, 100).cache() 之后，它似乎运行了更长的时间，但最终染上了 java.lang.OutOfMemoryError: Java heap space
如果你改变 spark.executor.memory=12g 运行时间会更长吗？您在系统上有多少内存要分配给工作人员？您可能还想在 conf/slaves 文件中添加更多工作人员。
嗨@javadba 看起来实际上是我试图在内存中组装整个排列数组，这正在杀死它。让combine 函数返回一个迭代器似乎可以解决问题。不过感谢您的宝贵时间！
关于使用 Spark Shell 2.2，我在“6g”上添加了-Dspark.executor.memory=6g，“error: Invalid literal number”。运行脚本sshell -i script.scala

【解决方案2】：

在文档 (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（下面的评论指的是这个）

【讨论】：

4G的10%不应该是410(M)左右吗？
是的，应该在 10% 左右，抱歉。所以：spark.yarn.executor.memoryOverhead=409
配置键 'spark.yarn.executor.memoryOverhead' 自 Spark 2.3 起已被弃用，将来可能会被删除。请改用新密钥“spark.executor.memoryOverhead”。

【解决方案3】：

正如已经建议的那样，调整内存可能是一个不错的方法，因为这是一项昂贵的操作，并且以一种丑陋的方式扩展。但也许一些代码更改会有所帮助。

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

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 转换为列表（或其他内容）。

【讨论】：

嗨@ohruunuruus，我认为滑动没有提供与我想要做的相同的行为。 toKey 函数只是将两个字符串组合在一起。
嗯好吧。当我可以访问 spark-shell 时，我将不得不看看。无论如何，您可能会探索执行组合操作的其他方式。
嗨@ohruunuruus，成功了，我认为返回迭代器才是真正的诀窍，因为我的for循环试图在内存中组装一个巨大的数组，但由于内存不足而失败。我知道我的 for 循环看起来不太像 scala，但我昨天开始学习，所以这就是我能做的。谢谢你的时间！

【解决方案4】：

在长期回归拟合期间我遇到了同样的问题。我缓存了训练集和测试集。它解决了我的问题。

train_df, test_df = df3.randomSplit([0.8, 0.2], seed=142)
pipeline_model = pipeline_object.fit(train_df)

pipeline_model 行给了java.lang.OutOfMemoryError: GC overhead limit exceeded 但是当我使用

train_df, test_df = df3.randomSplit([0.8, 0.2], seed=142)
train_df.cache()
test_df.cache()
pipeline_model = pipeline_object.fit(train_df)

成功了。

【讨论】：

【解决方案5】：

当我将 spark.memory.fraction 增加到大于 0.6 的值时，这个 JVM 垃圾收集错误重复发生。因此最好将值保留为默认值，以避免 JVM 垃圾收集错误。 https://forums.databricks.com/questions/2202/javalangoutofmemoryerror-gc-overhead-limit-exceede.html 也建议这样做。

有关为什么0.6 是spark.memory.fraction 的最佳价值的更多信息，请参阅https://issues.apache.org/jira/browse/SPARK-15796。

【讨论】：