Spark 1.5.2 随机/序列化 - 内存不足答案

【问题标题】：Spark 1.5.2 Shuffle/Serialization - running out of memorySpark 1.5.2 随机/序列化 - 内存不足
【发布时间】：2016-05-06 15:27:18
【问题描述】：

我正在处理数百 GB 数据集（大约 2B 行）。其中一项操作是将 RDD 或 scala 案例对象（包含双精度、映射、集合）减少为单个实体。最初我的操作正在执行groupByKey，但它很慢并且正在执行高 GC。所以我尝试将其转换为aggregateByKey，后来甚至转换为reduceByKey，希望避免我在使用groupBy时遇到的高用户内存分配、随机播放活动和高gc问题。

应用程序资源： 23GB exec mem + 4GB 开销。每个实例 20 个实例和 6 个核心。随机播放比例从 0.2 到 0.4

可用集群资源 10 个节点，yarn 总共 600GB，最大容器大小 32GB

2016-05-02 22:38:53,595 INFO [sparkDriver-akka.actor.default-dispatcher-14] org.apache.spark.MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 3 to hdn2.mycorp:45993
2016-05-02 22:38:53,832 INFO [sparkDriver-akka.actor.default-dispatcher-14] org.apache.spark.storage.BlockManagerInfo: Removed broadcast_4_piece0 on 10.250.70.117:52328 in memory (size: 2.1 KB, free: 15.5 MB)
2016-05-02 22:39:03,704 WARN [New I/O worker #5] org.jboss.netty.channel.DefaultChannelPipeline: An exception was thrown by a user handler while handling an exception event ([id: 0xa8147f0c, /10.250.70.110:48056 => /10.250.70.117:38300] EXCEPTION: java.lang.OutOfMemoryError: Java heap space)
java.lang.OutOfMemoryError: Java heap space
        at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
        at java.nio.ByteBuffer.allocate(ByteBuffer.java:331)
        at org.jboss.netty.buffer.CompositeChannelBuffer.toByteBuffer(CompositeChannelBuffer.java:649)
        at org.jboss.netty.buffer.AbstractChannelBuffer.toByteBuffer(AbstractChannelBuffer.java:530)
        at org.jboss.netty.channel.socket.nio.SocketSendBufferPool.acquire(SocketSendBufferPool.java:77)
        at org.jboss.netty.channel.socket.nio.SocketSendBufferPool.acquire(SocketSendBufferPool.java:46)
        at org.jboss.netty.channel.socket.nio.AbstractNioWorker.write0(AbstractNioWorker.java:194)
        at org.jboss.netty.channel.socket.nio.AbstractNioWorker.writeFromTaskLoop(AbstractNioWorker.java:152)
        at org.jboss.netty.channel.socket.nio.AbstractNioChannel$WriteTask.run(AbstractNioChannel.java:335)
        at org.jboss.netty.channel.socket.nio.AbstractNioSelector.processTaskQueue(AbstractNioSelector.java:366)
        at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:290)
        at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:90)
        at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:744)
2016-05-02 22:39:05,783 ERROR [sparkDriver-akka.actor.default-dispatcher-14] org.apache.spark.rpc.akka.ErrorMonitor: 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:843)
        at akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:843)
        at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
        at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:842)
        at akka.remote.EndpointWriter.writeSend(Endpoint.scala:743)
        at akka.remote.EndpointWriter$$anonfun$2.applyOrElse(Endpoint.scala:718)
        at akka.actor.Actor$class.aroundReceive(Actor.scala:467)
        at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:411)
        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:397)
        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)
2016-05-02 22:39:05,783 ERROR [sparkDriver-akka.actor.default-dispatcher-2] akka.actor.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)
                                                                                                                                                                                                                              67247,1       99%

关于工作 读取包含大约 20 个字段的输入数据集。 1B-2B。创建一个聚合超过 10 个唯一字段的输出数据集。这基本上成为查询条件。但是在这 10 个字段中，有 3 个字段表示它们的各种组合，因此我们不必查询多条记录来获取一组记录。在这 3 个字段中，让 sat a、b 和 c 各有 11、2 和 2 个可能的值。所以我们可以获得给定键的最大 2^11 -1 * 2^2 - 1 * 2^2 -1 组合。

//pseudo code where I use aggregateByKey 

case class UserDataSet(salary: Double, members: Int, clicks: Map[Int, Long],
    businesses: Map[Int, Set[Int]])...) //About 10 fileds with 5 of them are maps

def main() = {

      create combinationRDD of type (String, Set[Set]) Rdd from input dataset which represent all combination
      create a joinedRdd of type (String, UserDataSet) - where key at this point already a final key which contains 10 unique fields; value is a UserDataSet

//This is where things fails
     val finalDataSet = joinedRdd.aggregateByKey(UserDataSet.getInstance())(processDataSeq, processDataMerge)

}    

private def processDataMerge(map1: UserDataSet, map2: UserDataSet) = { 

    map1.clicks ++= map2.clicks (deep merge of course to avoid overwriting of map keys)
    map1.salary += map2.salary

    map1
}

【问题讨论】：

为了让我们为您提供帮助，我们至少需要查看一些代码来了解您对这 2B 行的实际操作。
刚刚添加了一些工作描述和伪代码。如果我应该提供更多信息，请告诉我。

标签： scala serialization apache-spark shuffle

【解决方案1】：

所以问题确实是驱动程序内存不足而不是执行程序。因此错误出现在驱动程序日志中。呃。但是从日志中不是很清楚。驱动程序用完了，因为 1）它使用默认值 -Xmx900m 2）Spark 驱动程序依赖于 akka 库，而 akka 库依赖于顽固的 JavaSerializer，它使用字节数组而不是流来序列化对象。作为临时解决方案，我将 spark.driver.memory 增加到 4096m，从那以后我没有看到内存错误。不过，感谢大家对问题空间的一些见解。

【讨论】：

【解决方案2】：

为了能够提供帮助，您应该发布代码并解释输入数据。

为什么是数据？在按键聚合时，为了实现最佳并行性并避免问题，重要的是要了解键分布的样子以及基数。

让我解释一下它们是什么以及为什么它们很重要。假设您按国家/地区汇总...地球上大约有 250 个国家/地区，因此键的基数约为 250。

基数很重要，因为低基数可能会扼杀您的并行性。例如，如果您 90% 的数据是针对美国的，并且您有 250 个节点，那么一个节点将处理 90% 的数据。

这就引出了分布的概念，也就是说，当你按键分组时，每个键有多少值就是你的值分布。为了获得最佳并行度，理想情况下，您希望每个键的值数量大致相同。

现在，如果您的数据的基数非常高，但值分布不是最优的，那么统计上的结果应该是平衡的。例如，假设您有 apache 日志，大多数用户只访问几个页面，但有些用户访问很多（就像机器人一样）。如果用户数量远大于您的节点数量，则拥有大量数据的用户会分布在节点周围，因此并行性不会受到影响。

当您使用低基数的键时，通常会出现问题。如果值的分布不好，它会导致问题不太可能是洗衣机不平衡。

最后但同样重要的是，它还很大程度上取决于您在 aggregateByKey 上所做的事情。如果您在处理的 map 或 reduce 阶段泄漏对象，则很容易耗尽内存。

【讨论】：

谢谢。我知道数据分布不均。与初始数据集相比，最终键的基数为 100:1。 1B输入组-10M输出组。那有意义吗？但我认为我应该在 shuffle 之前和 shuffle 之后比较操作失败的地方，对吧？你能解释为什么我在驱动程序日志中看到这些错误吗？是执行者还是驱动程序的错误？是由于给予执行者和/或驱动程序的整体内存还是火花洗牌块本身的限制？