spark out out memory错误

Question

我的spark程序在小数据集(大约400 to )上工作得很好。然而，当我将它扩展到大型数据集时。我开始收到错误java.lang.OutOfMemoryError: Java heap space或java.lang.OutOfMemoryError: Requested array size exceeds VM limit

我的程序如下所示: sc.textFile -> map -> filter -> groupBy -> saveAsObjectFile

groupBy生成类型为RDD[ (int，IteratableA )]的结果

出现saveAsObjectFile时出现错误。我能想到的唯一原因是:在groupBy步骤中，一些键包含了太多的数据。但是我用Hive检查了所有的密钥，最大的一个是330808。A类也不是很大。

我的配置是：-driver-memory 20G --num-executors 120 --executor-memory 30G Spark version: 1.4

15/07/03 07:05:06 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 java.util.Arrays.copyOf(Arrays.java:2271)
        at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:113)
        at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)
        at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:140)
        at java.io.ObjectOutputStream$BlockDataOutputStream.drain(ObjectOutputStream.java:1876)
        at java.io.ObjectOutputStream$BlockDataOutputStream.setBlockDataMode(ObjectOutputStream.java:1785)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1188)
        at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347)
        at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply$mcV$sp(Serializer.scala:129)
        at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply(Serializer.scala:129)
        at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply(Serializer.scala:129)
        at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
        at akka.serialization.JavaSerializer.toBinary(Serializer.scala:129)
        at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:36)
        at akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:845)
        at akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:845)
        at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
        at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:844)
        at akka.remote.EndpointWriter.writeSend(Endpoint.scala:747)
        at akka.remote.EndpointWriter$$anonfun$2.applyOrElse(Endpoint.scala:722)
        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)

Answer 1

驱动程序OutOfMemory的快速解决方案是使用“spark.driver.memory”属性增加驱动程序内存。

下面的文章可能有助于为驱动程序和执行器http://www.wdong.org/wordpress/blog/2015/01/08/spark-on-yarn-where-have-all-my-memory-gone/分配内存

还要注意的是，GroupByKey的操作成本更高。所以尽量避免使用reduceByKey。

http://databricks.gitbooks.io/databricks-spark-knowledge-base/content/best_practices/prefer_reducebykey_over_groupbykey.html

Answer 2

您的工作可能是不平衡的，因此一些分区会获得大量的键(及其值)。您可以尝试添加更多分区和/或编写一个自定义分区程序，根据您对数据的了解来平衡分区