SparkR 收集方法因 Java 堆 space 上的 OutOfMemory 而崩溃
SparkR collect method crashes with OutOfMemory on Java heap space
使用 SparkR，我正在尝试让 PoC 收集我从包含大约 400 万行的文本文件创建的 RDD。
我的 Spark 集群 运行ning 在 Google 云中，部署了 bdutil，由 1 个 master 和 2 个 worker 组成，每个 15gb RAM 和 4 个内核。我的 HDFS 存储库基于 Google Storage with gcs-connector 1.4.0。
每台机器上都安装了 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 提示后，我再次尝试 运行 收集方法，结果如下：
[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.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(Endpoint.scala:58)
at akka.remote.DefaultMessageDispatcher.payload(Endpoint.scala:58)
at akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
at akka.remote.EndpointReader$$anonfun$receive.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 中完成后收集从不 returns？
我 Googled 我拥有的每一条信息，但我没有运气找到解决方案。任何帮助或提示将不胜感激！
谢谢
这似乎是 Java 内存中对象表示效率低下与一些明显的长寿命对象引用的简单组合，这导致某些集合无法及时进行垃圾回收新的 collect() 调用就地覆盖旧的。
我尝试了一些选项，对于包含 ~4M 行的 256MB 示例文件，我确实重现了您的行为，第一次收集很好，但第二次使用 SPARK_MEM=1g 时出现 OOM。然后我改为设置 SPARK_MEM=4g，然后我可以根据需要多次按 ctrl+c 和 re-运行 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) 在这一点上 运行ning 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[] -> 字符串 -> 字符串[] -> ... -> class scala.collection.mutable.DefaultEntry -> class [Lscala.collection.mutable.HashEntry; -> class scala.collection.mutable.HashMap -> class 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确实没有挂。
使用 SparkR，我正在尝试让 PoC 收集我从包含大约 400 万行的文本文件创建的 RDD。
我的 Spark 集群 运行ning 在 Google 云中，部署了 bdutil，由 1 个 master 和 2 个 worker 组成，每个 15gb RAM 和 4 个内核。我的 HDFS 存储库基于 Google Storage with gcs-connector 1.4.0。 每台机器上都安装了 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 提示后，我再次尝试 运行 收集方法，结果如下：
[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.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(Endpoint.scala:58)
at akka.remote.DefaultMessageDispatcher.payload(Endpoint.scala:58)
at akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
at akka.remote.EndpointReader$$anonfun$receive.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 中完成后收集从不 returns？
我 Googled 我拥有的每一条信息，但我没有运气找到解决方案。任何帮助或提示将不胜感激！
谢谢
这似乎是 Java 内存中对象表示效率低下与一些明显的长寿命对象引用的简单组合，这导致某些集合无法及时进行垃圾回收新的 collect() 调用就地覆盖旧的。
我尝试了一些选项，对于包含 ~4M 行的 256MB 示例文件，我确实重现了您的行为，第一次收集很好，但第二次使用 SPARK_MEM=1g 时出现 OOM。然后我改为设置 SPARK_MEM=4g，然后我可以根据需要多次按 ctrl+c 和 re-运行 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) 在这一点上 运行ning 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[] -> 字符串 -> 字符串[] -> ... -> class scala.collection.mutable.DefaultEntry -> class [Lscala.collection.mutable.HashEntry; -> class scala.collection.mutable.HashMap -> class 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确实没有挂。