我们先看一下在 YARN　上运行　Spark 的一些背景。查看之前的博文：点击这里查看

Spark（以及YARN） 需要关心的两项主要的资源是 CPU 和 内存， 磁盘 和 IO 当然也影响着 Spark 的性能，但是不管是 Spark 还是 Yarn 目前都没法对他们做实时有效的管理。

在一个 Spark 应用中，每个 Spark executor 拥有固定个数的 core 以及固定大小的堆大小。core 的个数可以在执行 spark-submit 或者 pyspark 或者 spark-shell 时，通过参数 –executor-cores 指定，或者在 spark-defaults.conf 配置文件或者 SparkConf 对象中设置 spark.executor.cores 参数。同样地，堆的大小可以通过 –executor-memory 参数或者 spark.executor.memory 配置项。core 配置项控制一个 executor 中task的并发数。 –executor-cores 5 意味着每个executor 中最多同时可以有5个 task 运行。memory 参数影响 Spark 可以缓存的数据的大小，也就是在 groupaggregate 以及 join 操作时 shuffle 的数据结构的最大值。

–num-executors 命令行参数或者spark.executor.instances 配置项控制需要的 executor 个数。从 CDH 5.4/Spark 1.3 开始，你可以避免使用这个参数，只要你通过设置 spark.dynamicAllocation.enabled 参数打开 动态分配 。动态分配可以使的 Spark 的应用在有后续积压的在等待的 task 时请求 executor，并且在空闲时释放这些 executor。

同时 Spark 需求的资源如何跟 YARN 中可用的资源配合也是需要着重考虑的，YARN 相关的参数有：

• yarn.nodemanager.resource.memory-mb 控制在每个节点上 container 能够使用的最大内存；
• yarn.nodemanager.resource.cpu-vcores 控制在每个节点上 container 能够使用的最大core个数；

请求5个 core 会生成向 YARN 要5个虚拟core的请求。从 YARN　请求内存相对比较复杂因为以下的一些原因：

• –executor-memory/spark.executor.memory 控制 executor 的堆的大小，但是 JVM 本身也会占用一定的堆空间，比如内部的 String 或者直接 byte buffer，executor memory 的 spark.yarn.executor.memoryOverhead 属性决定向 YARN 请求的每个 executor 的内存大小，默认值为max(384, 0.7 * spark.executor.memory);
• YARN 可能会比请求的内存高一点，YARN 的 yarn.scheduler.minimum-allocation-mb 和 yarn.scheduler.increment-allocation-mb 属性控制请求的最小值和增加量。

下面展示的是 Spark on YARN 内存结构：

如果这些还不够决定Spark executor 个数，还有一些概念还需要考虑的：

• 应用的master，是一个非 executor 的容器，它拥有特殊的从 YARN 请求资源的能力，它自己本身所占的资源也需要被计算在内。在 yarn-client 模式下，它默认请求 1024MB 和 1个core。在 yarn-cluster 模式中，应用的 master 运行 driver，所以使用参数 –driver-memory 和 –driver-cores 配置它的资源常常很有用。
• 在 executor 执行的时候配置过大的 memory 经常会导致过长的GC延时，64G是推荐的一个 executor 内存大小的上限。
• 我们注意到 HDFS client 在大量并发线程是时性能问题。大概的估计是每个 executor 中最多5个并行的 task 就可以占满的写入带宽。
• 在运行微型 executor 时（比如只有一个core而且只有够执行一个task的内存）扔掉在一个JVM上同时运行多个task的好处。比如 broadcast 变量需要为每个 executor 复制一遍，这么多小executor会导致更多的数据拷贝。

为了让以上的这些更加具体一点，这里有一个实际使用过的配置的例子，可以完全用满整个集群的资源。假设一个集群有6个节点有NodeManager在上面运行，每个节点有16个core以及64GB的内存。那么 NodeManager的容量：yarn.nodemanager.resource.memory-mb 和 yarn.nodemanager.resource.cpu-vcores 可以设为 63 * 1024 = 64512 （MB） 和 15。我们避免使用 100% 的 YARN container 资源因为还要为 OS 和 hadoop 的 Daemon 留一部分资源。在上面的场景中，我们预留了1个core和1G的内存给这些进程。Cloudera Manager 会自动计算并且配置。

所以看起来我们最先想到的配置会是这样的：–num-executors 6 –executor-cores 15 –executor-memory 63G。但是这个配置可能无法达到我们的需求，因为：

- 63GB+ 的 executor memory 塞不进只有 63GB 容量的 NodeManager；
- 应用的 master 也需要占用一个core，意味着在某个节点上，没有15个core给 executor 使用；
- 15个core会影响 HDFS IO的吞吐量。
配置成 –num-executors 17 –executor-cores 5 –executor-memory 19G 可能会效果更好，因为：
- 这个配置会在每个节点上生成3个 executor，除了应用的master运行的机器，这台机器上只会运行2个 executor
- –executor-memory 被分成3份（63G/每个节点3个executor）=21。 21 * （1 – 0.07） ~ 19。

调试并发

我们知道 Spark 是一套数据并行处理的引擎。但是 Spark 并不是神奇得能够将所有计算并行化，它没办法从所有的并行化方案中找出最优的那个。每个 Spark stage 中包含若干个 task，每个 task 串行地处理数据。在调试 Spark 的job时，task的个数可能是决定程序性能的最重要的参数。

那么这个数字是由什么决定的呢？在之前的博文中介绍了 Spark 如何将 RDD 转换成一组 stage。task 的个数与 stage 中上一个 RDD 的 partition 个数相同。而一个 RDD 的 partition 个数与被它依赖的 RDD 的 partition 个数相同，除了以下的情况： coalesce transformation 可以创建一个具有更少 partition 个数的 RDD，union transformation 产出的 RDD的 partition 个数是它父 RDD 的 partition 个数之和， cartesian 返回的 RDD 的 partition 个数是它们的积。