估计 Scala Spark 作业所需的内存答案

【问题标题】：Estimating required memory for Scala Spark job估计 Scala Spark 作业所需的内存
【发布时间】：2015-04-13 07:22:51
【问题描述】：

我正在尝试了解 Spark 作业需要多少内存。

当我运行作业时，我收到异常：

15/02/12 12:01:08 INFO rdd.HadoopRDD: Input split: file:/c:/data/example.txt:20661+20661
15/02/12 12:01:08 INFO rdd.HadoopRDD: Input split: file:/c:/data/example.txt:61983+20661
15/02/12 12:01:09 INFO rdd.HadoopRDD: Input split: file:/c:/data/example.txt:0+20661
15/02/12 12:01:09 INFO rdd.HadoopRDD: Input split: file:/c:/data/example.txt:61983+20661
15/02/12 12:01:09 INFO rdd.HadoopRDD: Input split: file:/c:/data/example.txt:41322+20661
15/02/12 12:01:09 INFO rdd.HadoopRDD: Input split: file:/c:/data/example.txt:20661+20661
15/02/12 12:01:11 ERROR executor.Executor: Exception in task 2.0 in stage 0.0 (TID 2)
java.lang.OutOfMemoryError: Java heap space

更多带有“15/02/12 12:01:08 INFO rdd.HadoopRDD: Input split: file:/c:/data/example.txt:20661+20661”的消息被打印出来，为了简洁起见，这里只是截断它们。

我正在记录计算，经过大约 1'000'000 次计算后，我收到了上述异常。

完成作业所需的计算次数为 64'000'000

目前我使用的是 2GB 内存，所以这是否意味着在内存中运行此作业而无需任何进一步的代码更改 2GB * 64 = 128GB 或者这是预测所需内存的过于简单的方法？

每个拆分文件如“15/02/12 12:01:09 INFO rdd.HadoopRDD: Input split: file:/c:/data/example.txt:20661+20661”是如何生成的？这些没有添加到文件系统中，因为本地机器上不存在“file:/c:/data/example.txt:20661+20661”？

【问题讨论】：

在回答这个问题之前，我需要多少内存？你应该回答这个问题：什么是内存贪婪？至于本地文件，它们应该可供所有具有相同路径的工作人员使用（使用共享文件夹或复制）。 example.txt:61983+20661 描述文件example.txt 上的分区/拆分61983+2066
@GQuintana “它们应该对具有相同路径的所有工作人员可用”或者“广播变量，可用于在所有节点的内存中缓存值”来自 spark.apache.org/docs/1.2.0/programming-guide.html。什么决定了“61983+2066”这个值？
将 RDD 作为广播变量加载意味着能够在每个节点中加载整个 RDD。分区将文件分成块，而每个节点处理一些块。值“61983+2066”由分区数和文件大小决定：.textFile 方法中有一个minPartitions 参数。
@G Quintana 61983 != 分区数？
那我猜它的 61983 是一个字节长度。该值在 Hadoop 的 FileInputFormat 类中计算。

标签： scala apache-spark

【解决方案1】：

为了估计所需的内存量，我使用了这种方法：

使用http://code.google.com/p/memory-measurer/，如Calculate size of Object in Java所述

一旦设置可以使用下面的代码来估计 Scala 集合的大小，然后这将提供 Spark 应用程序所需内存的指示：

object ObjectSizeDriver extends Application {

  val toMeasure = List(1,2,3,4,5,6);

  println(ObjectGraphMeasurer.measure(toMeasure));
  println(MemoryMeasurer.measureBytes(toMeasure));

}

【讨论】：