r - SparkR collect 方法因 Java 堆空间上的 OutOfMemory 而崩溃

Question

使用 SparkR，我正在尝试让 PoC 收集我从包含大约 400 万行的文本文件创建的 RDD。

我的 Spark 集群在 Google Cloud 中运行，部署了 bdutil，由 1 个主节点和 2 个 worker 组成，每个 15gb RAM 和 4 个内核。我的 HDFS 存储库基于带有 gcs-connector 1.4.0 的 Google Storage。SparkR 安装在每台机器上，基本测试在小文件上进行。

这是我使用的脚本:

Sys.setenv("SPARK_MEM" = "1g")
sc <- sparkR.init("spark://xxxx:7077", sparkEnvir=list(spark.executor.memory="1g"))
lines <- textFile(sc, "gs://xxxx/dir/")
test <- collect(lines)

我第一次运行它时，它似乎工作正常，所有任务都成功运行，spark 的 ui 显示作业已完成，但我再也没有收到 R 提示:

15/06/04 13:36:59 WARN SparkConf: Setting 'spark.executor.extraClassPath' to ':/home/hadoop/hadoop-install/lib/gcs-connector-1.4.0-hadoop1.jar' as a work-around.
15/06/04 13:36:59 WARN SparkConf: Setting 'spark.driver.extraClassPath' to ':/home/hadoop/hadoop-install/lib/gcs-connector-1.4.0-hadoop1.jar' as a work-around.
15/06/04 13:36:59 INFO Slf4jLogger: Slf4jLogger started
15/06/04 13:37:00 INFO Server: jetty-8.y.z-SNAPSHOT
15/06/04 13:37:00 INFO AbstractConnector: Started SocketConnector@0.0.0.0:52439
15/06/04 13:37:00 INFO Server: jetty-8.y.z-SNAPSHOT
15/06/04 13:37:00 INFO AbstractConnector: Started SelectChannelConnector@0.0.0.0:4040

15/06/04 13:37:54 INFO GoogleHadoopFileSystemBase: GHFS version: 1.4.0-hadoop1
15/06/04 13:37:55 WARN LoadSnappy: Snappy native library is available
15/06/04 13:37:55 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/06/04 13:37:55 WARN LoadSnappy: Snappy native library not loaded
15/06/04 13:37:55 INFO FileInputFormat: Total input paths to process : 68
[Stage 0:=======================================================>                                                                                     (27 + 10) / 68]

然后在按 CTRL-C 返回 R 提示后，我尝试再次运行 collect 方法，结果如下:

[Stage 1:==========================================================>                                                                                   (28 + 9) / 68]15/06/04 13:42:08 ERROR ActorSystemImpl: Uncaught fatal error from thread [sparkDriver-akka.remote.default-remote-dispatcher-5] shutting down ActorSystem [sparkDriver]
java.lang.OutOfMemoryError: Java heap space
        at org.spark_project.protobuf.ByteString.toByteArray(ByteString.java:515)
        at akka.remote.serialization.MessageContainerSerializer.fromBinary(MessageContainerSerializer.scala:64)
        at akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104)
        at scala.util.Try$.apply(Try.scala:161)
        at akka.serialization.Serialization.deserialize(Serialization.scala:98)
        at akka.remote.MessageSerializer$.deserialize(MessageSerializer.scala:23)
        at akka.remote.DefaultMessageDispatcher.payload$lzycompute$1(Endpoint.scala:58)
        at akka.remote.DefaultMessageDispatcher.payload$1(Endpoint.scala:58)
        at akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
        at akka.remote.EndpointReader$$anonfun$receive$2.applyOrElse(Endpoint.scala:937)
        at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
        at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:415)
        at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
        at akka.actor.ActorCell.invoke(ActorCell.scala:487)
        at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)
        at akka.dispatch.Mailbox.run(Mailbox.scala:220)
        at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)
        at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
        at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
        at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
        at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

我理解异常消息，但我不明白为什么我第二次收到此消息。另外，为什么收集在 Spark 中完成后不再返回？

我用谷歌搜索了所有的信息，但我没有找到解决方案。任何帮助或提示将不胜感激!

谢谢

Answer 1

这看起来确实是 Java 内存中对象表示的简单组合，效率低下，加上一些明显的长生命周期对象引用，导致某些集合无法及时进行垃圾回收，以便新的 collect() 调用就地覆盖旧的。

我尝试了一些选项，对于包含 ~4M 行的 256MB 示例文件，我确实重现了您的行为，第一次收集很好，但第二次 OOM，当使用 SPARK_MEM=1g 时.然后我设置 SPARK_MEM=4g相反，然后我可以按 ctrl+c 并重新运行 test <- collect(lines)想多少次都行。

一方面，即使引用没有泄漏，请注意在您第一次运行 test <- collect(lines) 之后, 变量 test拿着那巨大的线条阵列，你第二次调用它时，collect(lines)在最终分配给 test 之前执行 变量，因此在任何直接的指令排序中，都无法对 test 的旧内容进行垃圾回收。 .这意味着第二次运行将使 SparkRBackend 进程同时持有整个集合的两个副本，从而导致您看到的 OOM。

为了诊断，我在主机上启动了 SparkR 并首先运行

dhuo@dhuo-sparkr-m:~$ jps | grep SparkRBackend
8709 SparkRBackend

我还检查了top它使用了大约 22MB 的内存。我用 jmap 获取了堆配置文件:

jmap -heap:format=b 8709
mv heap.bin heap0.bin

然后我跑了第一轮test <- collect(lines)此时运行 top使用 ~1.7g 的 RES 内存显示它。我捕获了另一个堆转储。最后我也试了test <- {}摆脱引用以允许垃圾收集。这样做之后，打印出 test并显示它是空的，我抓取另一个堆转储并注意到 RES 仍然显示 1.7g。我用了jhat heap0.bin分析原始heap dump，得到:

Heap Histogram

All Classes (excluding platform)

Class   Instance Count  Total Size
class [B    25126   14174163
class [C    19183   1576884
class [<other>  11841   1067424
class [Lscala.concurrent.forkjoin.ForkJoinTask; 16  1048832
class [I    1524    769384
...

运行收集后，我有:

Heap Histogram

All Classes (excluding platform)

Class   Instance Count  Total Size
class [C    2784858 579458804
class [B    27768   70519801
class java.lang.String  2782732 44523712
class [Ljava.lang.Object;   2567    22380840
class [I    1538    8460152
class [Lscala.concurrent.forkjoin.ForkJoinTask; 27  1769904

即使在我取消了 test 之后, 它保持大致相同。这向我们展示了 2784858 个 char[] 的实例，总大小为 579MB，还有 2782732 个 String 的实例，大概是在它上面放置那些 char[]。我一直跟着引用图往上走，得到了类似的东西

char[] -> String -> String[] -> ... -> class scala.collection.mutable.DefaultEntry -> class [Lscala.collection.mutable.HashEntry; -> 类 scala.collection.mutable.HashMap -> 类 edu.berkeley.cs.amplab.sparkr.JVMObjectTracker$ -> java.util.Vector@0x785b48cd8(36 字节) -> sun.misc.Launcher$AppClassLoader@0x7855c31a8( 138 字节)

然后 AppClassLoader 有几千个入站引用。因此，沿着这条链的某处，某些东西本应删除它们的引用但没有这样做，导致整个收集的数组在我们尝试获取它的第二个副本时位于内存中。

最后，回答你关于 collect 之后挂起的问题，它似乎与不适合 R 进程内存的数据有关；这是与该问题相关的主题:https://www.mail-archive.com/user@spark.apache.org/msg29155.html

我确认使用只有几行的较小文件，然后运行 ​​collect确实不挂。