SparkR收集方法在Java堆空間上使用OutOfMemory崩潰

Question

使用SparkR，我試圖讓PoC收集從包含大約4M行的文本文件創建的RDD。

我的Spark羣集在Google Cloud中運行，並且由bdutil進行部署，由1個主服務器和2個工作服務器組成，每個服務器有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) 

我第一次運行它，它似乎是工作的罰款，所有的任務都成功運行，火花的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文件，我確實重現了第一次收集很好時的行爲，但使用SPARK_MEM=1g時第二次收到了OOM。然後，我設置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的表明它使用〜RES內存1.7克。我抓起另一個堆轉儲。最後，我還嘗試test <- {}以擺脫允許垃圾收集的引用。這樣做後，打印出test並顯示它是空的，我抓起另一個堆轉儲，並注意到RES仍然顯示1.7克。我以前jhat heap0.bin來分析原始堆轉儲，並得到了：

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，它基本保持不變。這向我們展示了char []的2784858個實例，總大小爲579MB，並且還顯示了2782732個String實例，大概在它上面保存了這些char []。我跟着參考圖形一直向上，並得到類似於

char [] - > String - > String [] - > ... - > 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 bytes） - > sun.misc.Launcher$AppClassLoader@0x7855c31a8（ 138字節）

然後AppClassLoader有類似成千上萬的入站引用。所以在這條鏈的某個地方，應該刪除它們的引用，但是沒有這樣做，導致整個收集的數組駐留在內存中，而我們試圖獲取它的第二個副本。

最後，要回答關於在collect之後掛起的問題，看起來它與R進程內存中不適合的數據有關;這裏是一個與此問題有關的線程：https://www.mail-archive.com/user@spark.apache.org/msg29155.html

我確認使用只有少數幾行的較小文件，然後運行collect確實不會掛起。