java - Spark OutOfMemory 小文本数据错误

Question

我正在努力实现一种算法，并在本地节点上的 Spark(Scala 接口(interface))中对中型数据进行测试。我从非常简单的处理开始，我得到了 java.lang.OutOfMemoryError: Java heap space，尽管我很确定数据不够大，这样的错误是合理的。这是最少的中断代码:

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkContext, SparkConf}

val conf = new SparkConf()
  .setMaster("local[4]")
  .setAppName("AdultProcessing")
  .set("spark.executor.memory", "1g")
val sc = new SparkContext(conf)

val dataFile = "data/census/processed/census-income.data"
val censusData: RDD[String] = sc.textFile(dataFile, 4)
val censusDataPreprocessed = censusData.map { row =>
  val separators: Array[Char] = ":,".toCharArray
  row.split(separators)
}

val res = censusDataPreprocessed.collect()

我使用的数据是 the classic census data , 未压缩。它有 100MB 和近 20 万行。我机器上的内存量应该绰绰有余:

nietaki@xebab$ free -tm
             total       used       free     shared    buffers     cached
Mem:         15495      12565       2929          0        645       5608
-/+ buffers/cache:       6311       9183
Swap:         3858          0       3858
Total:       19354      12566       6788

每个虚拟节点的数据文件 block 都不到 30MB，我正在执行的唯一处理是将行字符串拆分为少于 50 个项目的数组。我无法相信仅此操作就会耗尽内存。

在尝试调试情况时，我发现将节点数减少到 1，或者增加 SparkContext.textFile() 的 minPartitions例如，从 4 到 8 的争论解决了这个问题，但它并没有让我变得更聪明。

我使用的是 Spark 1.0.0 和 Scala 2.10.4。我直接从 sbt 启动项目:sbt run -Xmx2g -Xms2g。

Answer 1

JVM 非常耗费内存。 Spark 在 JVM 上运行。

我建议您使用分析器检查堆，以找出您的记录实际使用的内存。在我的例子中，它们是“静止”大小的 2 倍，并且它们是原始类型和字符串的组合。

在您的情况下，字符串尤其是内存吞噬者。 ""(空字符串)长约 40 个字节 - 更长的字符串抵消了结构的成本。见[1]

将上一个资源中的表应用于您的数据:

line: String = 73, Not in universe, 0, 0, High school graduate, 0, Not in universe, Widowed, Not in universe or children, Not in universe, White, All other, Female, Not in universe, Not in universe, Not in labor force, 0, 0, 0, Nonfiler, Not in universe, Not in universe, Other Rel 18+ ever marr not in subfamily, Other relative of householder, 1700.09, ?, ?, ?, Not in universe under 1 year old, ?, 0, Not in universe, United-States, United-States, United-States, Native- Born in the United States, 0, Not in universe, 2, 0, 95, - 50000.

line.size
Int = 523
def size(s:String) = s.size/4*8+40
line.split(",.").map(w=>size(w)).sum
Int = 2432

因此，由于所有这些小字符串，内存中的数据大约是“静态”大小的 5 倍。尽管如此，20 万行数据仍占大约 500MB。这可能表明您的执行程序正在以 512MB 的默认值运行。尝试将 'spark.executor.memory' 设置为更高的值，但也考虑堆大小 >8Gb 以舒适地使用 Spark。