apache-spark - sc.textfile 与 sc.wholeTextFiles + flatMapValues 的内存使用情况

Question

我有一组我想读入 RDD 的日志文件。
这些日志文件都是压缩的 gzip 文件，文件名是
日期盖章。

我一直在用sc.wholeTextFiles()读入文件，似乎我遇到了 Java 堆内存问题。为了隔离问题，我决定在一台机器上针对单个文件运行它作为测试用例。

我从这里获得了文件:

http://dumps.wikimedia.org/other/pagecounts-raw/

以下是文件的大小，包括压缩版本和未压缩版本:

 myuser@fembuntu$ ls -ltr pagecounts-20090505-180000*
 -rw-rw-r-- 1 myuser myuser  65170192 Sep 20  2009 pagecounts-20090505-180000.gz
-rw-rw-r-- 1 myuser myuser 233007266 Jan 22 00:23 pagecounts-20090505-180000.txt

并且机器上的可用内存如下:

myuser@fembuntu:~$ free -tm

       total       used       free     shared    buffers     cached
Mem:    4856       3018       1838        123         27        407
-/+ buffers/cache: 2583       2273
 Swap:  5080        849       4231
Total:  9937       3867       6069

所以我启动了 spark-shell，给了 executor 2G 的内存:

$ spark-shell --executor-memory 2G

scala> val pc_loc = "file:///home/myuser/data/pagecounts"
scala> val filename="/pagecounts-20090505-180000.gz"
filename: String = /pagecounts-20090505-180000.gz

在这里，我通过 sc.textFile() 读入数据并显示第一 2 行:

scala>  var rdd=sc.textFile(pc_loc + filename)
rdd: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[1] at textFile at <console>:31

scala> rdd.take(2)
res0: Array[String] = Array(aa.b Help:Books 1 5263, aa.b Main_Page 1 5416)

这很好用。

这里我使用 sc.wholeTextFiles() ，并通过 flatMapValues() 在新行上拆分获得一对RDD，其中行是键值对。这些值对应于使用 sc.textFile() 获得的 RDD 中的行.关键是文件路径。

scala> val pair_rdd=sc.wholeTextFiles(pc_loc + filename).flatMapValues(y => y.split("\n"))
pair_rdd: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[4] at flatMapValues at <console>:31

但是在执行操作时出现堆错误:

scala> pair_rdd.take(2)
16/01/22 01:13:36 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 1)
java.lang.OutOfMemoryError: Java heap space
at java.nio.HeapCharBuffer.<init>(HeapCharBuffer.java:57)
at java.nio.CharBuffer.allocate(CharBuffer.java:335)
at java.nio.charset.CharsetDecoder.decode(CharsetDecoder.java:795)
at org.apache.hadoop.io.Text.decode(Text.java:412)

谁能解释一下这里发生了什么？为什么 flatMapValues拆分行的调用似乎将 Java 堆内存使用量从水中吹出，从而导致堆错误？

Answer 1

您遇到的问题并非真正特定于 textFile对比 wholeTextFiles与 flatMapValues设想。看起来您的程序甚至还没有达到数据变平的程度，我很确定您在调用 count 时会遇到相同的异常。而不是 mapValues .

实际上只是创建大对象的问题。请记住 wholeTextFiles必须一次读取文件的完整内容，不能部分溢出到磁盘或部分垃圾收集。虽然 200MB 左右，但它并不是特别令人印象深刻，但由单个对象处理却是相当多的。此外，它必须驻留在一台机器上，这意味着分配负载要困难得多。

不像 wholeTextFiles , textFile在这种特殊情况下提供了更高的粒度。单个对象必须处理的数据要少得多，如果不再需要，可以有效地进行垃圾收集。

忽略对象的大小，看起来您在本地模式下使用 Spark。这意味着一切都由单个 JVM 处理。由于堆由所有线程共享，这意味着可用于实际处理的内存量可能低于您的预期。

最后你应该记住，只有一部分可用内存是为堆保留的。见 Garbage Collector Ergonomics和 How is the default java heap size determined? .如果你必须处理大对象，你总是可以使用 -Xms 覆盖默认的初始和最大堆大小。/-Xmx Java 选项。