Spark Streaming 2.0 GC 错误（随机播放问题）

【问题标题】：Spark Streaming 2.0 GC Error (Shuffle Issue)Spark Streaming 2.0 GC 错误（随机播放问题）
【发布时间】：2017-01-22 21:26:10
【问题描述】：

我的 kafka 主题中有数百万条消息，不幸的是，无论我得到多少时间限制，Spark Streaming 似乎都会将所有消息排队。我已经将我的独立服务器设置为 16 核和 64GB 内存，我给了我的驱动程序 12G 和 Executor 12G 内存。

16/09/14 19:27:45 WARN TaskSetManager: Lost task 4.0 in stage 1.0 (TID 7, 10.206.41.172): java.lang.OutOfMemoryError: GC overhead limit exceeded
        at com.twitter.chill.Tuple4Serializer.read(TupleSerializers.scala:68)
        at com.twitter.chill.Tuple4Serializer.read(TupleSerializers.scala:59)
        at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:708)
        at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.read(DefaultArraySerializers.java:396)
        at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.read(DefaultArraySerializers.java:307)
        at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:708)
        at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
        at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:551)
        at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:790)
        at org.apache.spark.serializer.KryoDeserializationStream.readObject(KryoSerializer.scala:229)
        at org.apache.spark.serializer.DeserializationStream.readValue(Serializer.scala:159)
        at org.apache.spark.util.collection.ExternalAppendOnlyMap$DiskMapIterator.readNextItem(ExternalAppendOnlyMap.scala:515)
        at org.apache.spark.util.collection.ExternalAppendOnlyMap$DiskMapIterator.hasNext(ExternalAppendOnlyMap.scala:535)
        at scala.collection.Iterator$$anon$1.hasNext(Iterator.scala:1004)
        at org.apache.spark.util.collection.ExternalAppendOnlyMap$ExternalIterator.org$apache$spark$util$collection$ExternalAppendOnlyMap$ExternalIterator$$readNextHashCode(ExternalAppendOnlyMap.scala:332)
        at org.apache.spark.util.collection.ExternalAppendOnlyMap$ExternalIterator$$anonfun$5.apply(ExternalAppendOnlyMap.scala:316)
        at org.apache.spark.util.collection.ExternalAppendOnlyMap$ExternalIterator$$anonfun$5.apply(ExternalAppendOnlyMap.scala:314)
        at scala.collection.immutable.List.foreach(List.scala:381)
        at org.apache.spark.util.collection.ExternalAppendOnlyMap$ExternalIterator.<init>(ExternalAppendOnlyMap.scala:314)
        at org.apache.spark.util.collection.ExternalAppendOnlyMap.iterator(ExternalAppendOnlyMap.scala:288)
        at org.apache.spark.Aggregator.combineValuesByKey(Aggregator.scala:43)
        at org.apache.spark.shuffle.BlockStoreShuffleReader.read(BlockStoreShuffleReader.scala:91)
        at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:109)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
        at org.apache.spark.scheduler.Task.run(Task.scala:85)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)


org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 0
        at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:695)
        at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:691)
        at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
        at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
        at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
        at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
        at org.apache.spark.MapOutputTracker$.org$apache$spark$MapOutputTracker$$convertMapStatuses(MapOutputTracker.scala:691)
        at org.apache.spark.MapOutputTracker.getMapSizesByExecutorId(MapOutputTracker.scala:145)
        at org.apache.spark.shuffle.BlockStoreShuffleReader.read(BlockStoreShuffleReader.scala:49)
        at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:109)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
        at org.apache.spark.scheduler.Task.run(Task.scala:85)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)


        org.apache.spark.shuffle.FetchFailedException: /tmp/spark-e4238a07-bf89-4a7d-9de3-176cba0a076d/executor-93a11b25-1cb1-4e13-b1b6-b1d64d3a9602/blockmgr-c11cd046-7c37-429c-9137-936d391d3cbc/30/shuffle_0_0_0.index (No such file or directory)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:357)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:332)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:54)
        at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
        at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:434)
        at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:32)
        at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39)
        at org.apache.spark.util.collection.ExternalAppendOnlyMap.insertAll(ExternalAppendOnlyMap.scala:154)
        at org.apache.spark.Aggregator.combineValuesByKey(Aggregator.scala:41)
        at org.apache.spark.shuffle.BlockStoreShuffleReader.read(BlockStoreShuffleReader.scala:91)
        at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:109)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
        at org.apache.spark.scheduler.Task.run(Task.scala:85)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.io.FileNotFoundException: /tmp/spark-e4238a07-bf89-4a7d-9de3-176cba0a076d/executor-93a11b25-1cb1-4e13-b1b6-b1d64d3a9602/blockmgr-c11cd046-7c37-429c-9137-936d391d3cbc/30/shuffle_0_0_0.index (No such file or directory)
        at java.io.FileInputStream.open0(Native Method)
        at java.io.FileInputStream.open(FileInputStream.java:195)
        at java.io.FileInputStream.<init>(FileInputStream.java:138)
        at org.apache.spark.shuffle.IndexShuffleBlockResolver.getBlockData(IndexShuffleBlockResolver.scala:192)
        at org.apache.spark.storage.BlockManager.getBlockData(BlockManager.scala:278)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.fetchLocalBlocks(ShuffleBlockFetcherIterator.scala:258)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.initialize(ShuffleBlockFetcherIterator.scala:292)
        at org.apache.spark.storage.ShuffleBlockFetcherIterator.<init>(ShuffleBlockFetcherIterator.scala:120)
        at org.apache.spark.shuffle.BlockStoreShuffleReader.read(BlockStoreShuffleReader.scala:45)
        ... 9 more


java.io.IOException: Connection reset by peer
        at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
        at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
        at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
        at sun.nio.ch.IOUtil.read(IOUtil.java:192)
        at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
        at io.netty.buffer.PooledUnsafeDirectByteBuf.setBytes(PooledUnsafeDirectByteBuf.java:313)
        at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:881)
        at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:242)
        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:119)
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
        at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
        at java.lang.Thread.run(Thread.java:745)

关于我能做些什么来处理这个问题有什么建议吗？我的解决方案是否只是为其提供更多内存？我指出一个洗牌问题，因为一个执行者有：

随机读取大小/记录：815.5 MB / 27445419

随机溢出（内存）：13GB

随机播放溢出（磁盘）：697.4 MB

我不知道它可能会洗牌。 

val messageStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, Int, Long, String)](ssc, getKafkaBrokers(), getKafkaTopics("raw"), (mmd: MessageAndMetadata[String, String]) => {
      (mmd.topic, mmd.partition, mmd.offset, mmd.message)
    })

    //first step is to take our RDD of messages and group them by the topic (x._1) and partition (x._2)
    messageStream.foreachRDD(x => x.groupBy(x => (x._1, x._2)).foreach(x => {
      //Populate parameters based on the grouping and get the instruction sets to run on each message
      val rawTopic: String = x._1._1
      val partitionID: Int = x._1._2
      val sourceSystemName: String = rawTopic.split("_")(0)
      val cleanTopic = sourceSystemName + "_clean"
      val topicID = getTopicID(rawTopic)
      val schemaID = getPartitionByTopic(topicID).filter(x => x._1 == partitionID)(0)._2
      val classpath = getClasspath(schemaID.toInt)
      val classPath = bcClassMap.value.get(classpath)
      val cleanerObj = classPath.newInstance()
      val cleanMethod = classPath.getMethod("clean", Class.forName("java.lang.String"))

      //For each message run the instruction sets we've gotten above
      x._2.foreach(kafkaMessage => {
        val offset: Long = kafkaMessage._3
        val rawRecord: String = kafkaMessage._4
        val cleanRecord = cleanMethod.invoke(cleanerObj, rawRecord).asInstanceOf[String]
        if (cleanRecord != null) {
          sendKafka(cleanRecord, cleanTopic, partitionID.toString)
        }
        offsetUpdate(schemaID, offset.toInt, topicID)
      })
    }))

【问题讨论】：

  • 在代码中添加。我有很多内置函数，但很乐意详细说明其中的任何一个。
  • 我给了驱动程序 30GB 的内存，似乎没有崩溃，但仍然担心溢出。
  • 这groupBy 闻起来很臭。有什么可以真正证明它的合理性吗？既然看起来您无论如何都单独处理单个消息，那有什么意义呢？
  • 之所以拥有它，是因为它降低了对 getTopicID、getPartitionByTopic、getClasspath 的调用量，因此对于每组消息，我们只调用一次。而不是为每条消息调用它。我不想毁掉我的 API 服务器。

标签： scala apache-spark spark-streaming rdd


【解决方案1】：

这里的一个明显嫌疑人是最初的groupBy。

虽然它可以在批处理应用程序中进行维护，但当您对数据分布有所了解时，它在流应用程序中是不可接受的，其中单个倾斜批处理可能会破坏整个管道。即使它没有达到那么高，它也会削弱整体性能。

由于您的代码并不真正依赖于分组，而这只是一种优化尝试，因此您的情况非常好。例如，您可以：

  • 完全删除groupBy。
  • 为每个分区或执行程序使用共享缓存来处理昂贵的 API 调用。

如果 API 调用的数量仍然很高，您可以生成加盐分区键：

_.keyBy(x => (x._1, x._2, smallRandomInteger))

并对每个 RDD 重新分区。

您也可以对groupBy 使用加盐，但如果这样使用，您只能解决数据倾斜问题，而不是分组和维护大型本地缓冲区的成本。

【讨论】：

  • 什么意思：“每个分区或执行程序的共享缓存”
  • 一个对象，它将充当外部 API 的代理，并为不同的方法调用维护缓存。如果响应随着时间的推移是恒定的，则每个执行程序都可以处理（单例对象应该就可以了），否则您可以使用mapPartitions 并在那里初始化缓存。
  • 只使用 .map(x => ((x._1, x._2), (x._3, x._4))) 会比重新分区便宜吗？我觉得 keyBy 会添加可能以这种格式完成的额外信息？
  • 会更便宜。我虽然出于某种原因您更喜欢这种格式，这就是您使用 groupBy 的原因。
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2023-01-23
  • 2017-03-28
  • 2016-09-10
  • 2016-01-20
  • 2017-07-02
  • 1970-01-01
相关资源
最近更新 更多
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode