hadoop - YARN 上的 Spark 应用程序的物理内存使用量不断增加

Question

我在 YARN 客户端模式下运行一个 Spark 应用程序，有六个执行器(每个四个内核和执行器内存 = 6 GB，开销 = 4 GB，Spark 版本:1.6.3/2.1.0)。

我发现我的执行程序内存一直在增加，直到被节点管理器杀死；它给出了告诉我提升 spark.yarn.excutor.memoryOverhead 的信息。

我知道这个参数主要是控制堆外分配内存的大小。但是我不知道Spark引擎会在什么时候以及如何使用这部分内存。另外增加那部分内存并不总能解决我的问题。有时有效，有时无效。当输入数据很大时，它趋向于无用。

仅供引用，我的应用程序的逻辑非常简单。意思是把一天(一天一个目录)产生的小文件合并成一个，写回HDFS。这是核心代码:

val df = spark.read.parquet(originpath)
              .filter(s"m = ${ts.month} AND d = ${ts.day}")
              .coalesce(400)
val dropDF = df.drop("hh").drop("mm").drop("mode").drop("y").drop("m").drop("d")

dropDF.repartition(1).write
      .mode(SaveMode.ErrorIfExists)
      .parquet(targetpath)

源文件可能有几百到几千级的分区。 parquet 文件的总大小约为 1 到 5 GB。

另外我发现在shuffle从不同机器读取数据的步骤中，shuffle read的大小大约是输入大小的四倍，这是有线的还是一些我不知道的原理。

无论如何，我已经自己搜索了这个问题。有文章说在direct buffer memory上(我自己没设置)。

一些文章说人们通过更频繁的 full GC 来解决它。

此外，我在 Stack Overflow 上发现一个人的情况非常相似: Ever increasing physical memory for a Spark application in YARN

这个人声称这是 parquet 的一个错误，但有评论质疑他。此邮件列表中的人也可能在几个小时前收到一封来自 blondowski 的电子邮件，他在编写 JSON 时描述了这个问题: Executors - running out of memory

所以这看起来像是不同输出格式的常见问题。

希望有这方面经验的大侠给个解释。为什么会发生这种情况，解决这个问题的可靠方法是什么？

Answer 1

这几天正好和同事一起做一些调查。这是我的想法:从 spark 1.2 开始，我们使用带有堆外内存的 Netty 来减少 shuffle 和缓存 block 传输期间的 GC。就我而言，如果我尝试将内存开销增加得足够大。我将得到 Max 直接缓冲区异常。 Netty做 block 传输时，默认会有5个线程抓取数据 block 给目标执行器。在我的情况下，单个 block 太大而无法放入缓冲区。所以 gc 在这里无济于事。我最终的解决方案是在 repartition(1) 之前再做一次 repartition。只是为了制作比原来多 10 倍的分区。通过这种方式，我可以减少每个 chunk Netty 传输的大小。就这样我终于成功了。

另外我想说的是，将大数据集重新分区为单个文件并不是一个好的选择。这种极度不平衡的情况有点浪费您的计算资源。

欢迎大家评论，这部分我还不是很明白。