常规性能调优一：最优资源配置

Spark 性能调优的第一步，就是为任务分配更多的资源，在一定范围内，增加资源的分配与性能的提升是成正比的， 实现了最优的资源配置后， 在此基础上再考虑进行后面论述的性能调优策略。
资源的分配在使用脚本提交 Spark 任务时进行指定， 标准的 Spark 任务提交脚本如代码清单 2-1 所示：

/usr/opt/modules/spark/bin/spark-submit \
--class com.atguigu.spark.Analysis \
--num-executors 80 \
--driver-memory 6g \
--executor-memory 6g \
--executor-cores 3 \
/usr/opt/modules/spark/jar/spark.jar \

可以进行分配的资源如表 2-1 所示：
表 2-1 可分配资源表

对于具体资源的分配，我们分别讨论 Spark 的两种 Cluster 运行模式：
第一种是 Spark Standalone 模式，你在提交任务前，一定知道或者可以从运维部门获取到你可以使用的资源情况，在编写 submit 脚本的时候，就根据可用的资源情况进行资源的分配，比如说集群有 15 台机器，每台机器为 8G 内存，2 个 CPU core， 那么就指定 15 个 Executor， 每个 Executor 分配 8G 内存， 2 个 CPU core。

第二种是 Spark Yarn 模式，由于 Yarn 使用资源队列进行资源的分配和调度，在编写 submit 脚本的时候，就根据 Spark 作业要提交到的资源队列，进行资源的分配， 比如资源队列有 400G 内存，100 个 CPU core，那么指定 50 个 Executor，每个 Executor分配 8G 内存， 2 个 CPU core。

对表 2-1 中的各项资源进行了调节后， 得到的性能提升如表 2-2 所示：
表 2-2 资源调节后的性能提升

补充：生产环境 Spark submit 脚本配置
参数配置参考值：
--num-executors： 50~100
--driver-memory： 1G~5G
--executor-memory： 6G~10G
--executor-cores： 3
--master：实际生产环境一定使用 yarn-cluster

常规性能调优二：RDD 优化

• RDD 复用
  

  在对 RDD 进行算子时， 要避免相同的算子和计算逻辑之下对 RDD 进行重复的计算，如图 2-1 所示：
  image.png
  对图 2-1 中的 RDD 计算架构进行修改， 得到如图 2-2 所示的优化结果：
  image.png
• RDD 持久化
  在 Spark 中，当多次对同一个 RDD 执行算子操作时， 每一次都会对这个 RDD 以之前的父 RDD 重新计算一次，这种情况是必须要避免的， 对同一个 RDD 的重复计算是对资源的极大浪费，因此，必须对多次使用的 RDD 进行持久化，通过持久化将公共 RDD 的数据缓存到内存/磁盘中， 之后对于公共 RDD 的计算都会从内存/磁盘中直接获取 RDD 数据。

对于 RDD 的持久化， 有两点需要说明：
第一， RDD 的持久化是可以进行序列化的，当内存无法将 RDD 的数据完整的进行存放的时候，可以考虑使用序列化的方式减小数据体积， 将数据完整存储在内存中。
第二， 如果对于数据的可靠性要求很高， 并且内存充足， 可以使用副本机制， 对 RDD 数据进行持久化。当持久化启用了复本机制时，对于持久化的每个数据单元都存储一个副本，放在其他节点上面，由此实现数据的容错，一旦一个副本数据丢失，不需要重新计算，还可以使用另外一个副本。
RDD 尽可能早的 filter 操作
获取到初始 RDD 后，应该考虑尽早地过滤掉不需要的数据，进而减少对内存的占用，从而提升 Spark 作业的运行效率。

常规性能调优三：并行度调节

Spark 作业中的并行度指各个 stage 的 task 的数量。
如果并行度设置不合理而导致并行度过低， 会导致资源的极大浪费， 例如， 20 个 Executor，每个 Executor 分配 3 个 CPU core，而 Spark 作业有 40 个 task，这样每个 Executor 分配到的 task 个数是 2 个， 这就使得每个 Executor 有一个 CPU core 空闲， 导致资源的浪费。
理想的并行度设置， 应该是让并行度与资源相匹配， 简单来说就是在资源允许的前提下，并行度要设置的尽可能大， 达到可以充分利用集群资源。合理的设置并行度， 可以提升整个 Spark 作业的性能和运行速度。
Spark 官方推荐， task 数量应该设置为 Spark 作业总 CPU core 数量的 2~3 倍。之所以没有推荐 task 数量与 CPU core 总数相等， 是因为 task 的执行时间不同， 有的 task 执行速度快而有的 task 执行速度慢， 如果 task 数量与 CPU core 总数相等， 那么执行快的 task 执行完成后， 会出现 CPU core 空闲的情况。如果 task 数量设置为 CPU core 总数的 2~3 倍，那么一个 task 执行完毕后， CPU core 会立刻执行下一个 task， 降低了资源的浪费， 同时提升了 Spark 作业运行的效率。
Spark 作业并行度的设置如代码清单 2-2 所示：
代码清单 2-2 Spark作业并行度设置

常规性能调优四：广播大变量

默认情况下，task 中的算子中如果使用了外部的变量，每个 task 都会获取一份变量的复本，这就造成了内存的极大消耗。一方面，如果后续对 RDD 进行持久化，可能就无法将 RDD 数据存入内存， 只能写入磁盘， 磁盘 IO 将会严重消耗性能； 另一方面， task 在创建对象的时候， 也许会发现堆内存无法存放新创建的对象， 这就会导致频繁的 GC， GC 会导致工作线程停止，进而导致 Spark 暂停工作一段时间， 严重影响 Spark 性能。
假设当前任务配置了 20 个 Executor，指定 500 个 task，有一个 20M 的变量被所有 task 共用，此时会在 500 个 task 中产生 500 个副本，耗费集群 10G 的内存，如果使用了广播变量， 那么每个 Executor 保存一个副本，一共消耗 400M 内存，内存消耗减少了 5 倍。

广播变量在每个 Executor 保存一个副本，此 Executor 的所有 task 共用此广播变量，这让变量产生的副本数量大大减少。

在初始阶段，广播变量只在 Driver 中有一份副本。task 在运行的时候，想要使用广播变量中的数据，此时首先会在自己本地的 Executor 对应的 BlockManager 中尝试获取变量，如果本地没有，BlockManager 就会从 Driver 或者其他节点的 BlockManager上远程拉取变量的复本，并由本地的 BlockManager 进行管理；之后此 Executor 的所有 task 都会直接从本地的 BlockManager 中获取变量。

常规性能调优五：Kryo 序列化

默认情况下， Spark 使用 Java 的序列化机制。Java 的序列化机制使用方便， 不需要额外的配置，在算子中使用的变量实现 Serializable 接口即可，但是，Java 序列化机制的效率不高，序列化速度慢并且序列化后的数据所占用的空间依然较大。
Kryo 序列化机制比 Java 序列化机制性能提高 10 倍左右， Spark 之所以没有默认使用 Kryo 作为序列化类库， 是因为它不支持所有对象的序列化，同时 Kryo 需要用户在使用前注册需要序列化的类型，不够方便，但从 Spark 2.0.0 版本开始，简单类型、简单类型数组、字符串类型的 Shuffling RDDs 已经默认使用 Kryo 序列化方式了。
Kryo 序列化注册方式的实例代码如代码清单 2-3 所示：

代码清单 2-3 Kryo序列化机制配置代码
public class MyKryoRegistrator implements KryoRegistrator
{
@Override
public void registerClasses(Kryo kryo)
{
kryo.register(StartupReportLogs.class);
}
}

配置 Kryo 序列化方式的实例代码如代码清单 2-4 所示：

代码清单 2-4 Kryo序列化机制配置代码
//创建 SparkConf 对象
val conf = new SparkConf().setMaster(…).setAppName(…)
//使用 Kryo 序列化库，如果要使用 Java 序列化库，需要把该行屏蔽掉
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
//在 Kryo 序列化库中注册自定义的类集合，如果要使用 Java 序列化库，需要把该行屏蔽掉
conf.set("spark.kryo.registrator", "atguigu.com.MyKryoRegistrator");

1、Kryo类库如何使用？

• 优化缓存大小
  如果注册的要序列化的自定义的类型，本身特别大，比如超过100个field。那么就会导致要序列化的对象过大。此时就需要对Kryo本身进行优化。因为Kryo内部的缓存可能不够存放那么大的class对象。此时就需要调用SparkConf.set()方法设置
  spark.kryoserializer.buffer.mb
  参数的值，将其调大。默认情况下它的值是2，就是说最大缓存2M的对象，然后进行序列化。可以在必要时将其调大。比如设置为10.将2设置为10提高缓存，从而提高容纳的对象
• 预先注册自定义类型，虽然不注册自定义类型，Kryo类库也能正常工作，但是那样的话，对于要序列化的每个对象，都会保存一份它的全限定类名。此时反而会消耗大量内存。因此通常都建议预先注册好要序列化的自定义的类。

2、如何使用Kryo序列化机制？
如果要使用Kryo序列化机制，首先要用SparkConf设置一个参数，使用

newSparkConf().set(“spark.serializer”,”org.apache.serializer.KryoSerializer”)

即可，即将Spark的序列化器设置为KryoSerializer。这样，Spark在内部的一些操作，比如Shuffle，进行序列化时，就会使用Kryo类库进行高性能、快速、更低内存占用量的序列化了。
使用Kryo时，它要求是需要序列化的类，是要预先进行注册的，以获得最佳性能，如果不注册的话，那么Kryo必须时刻保存类型的全限定名，反而占用不少内存。Spark默认是对Scala中常用的类型自动注册了Kryo的，都在AllScalaRegistry类中。
但是，比如自己的算子中，使用了外部的自定义类型的对象，那么还是需要将其进行注册。

spark性能优化之高性能序列化类库

1、Spark提供的两种序列化机制是什么？
答：Java序列化机制：默认情况下，Spark使用Java自身的ObjectInputStream和ObjectOutPutStream机制进行对象的序列化。只要你的类实现了serializable接口，那么都是可以序列化的。而且Java序列化机制是提供了自定义序列化支持的，只要你实现了Externalizable接口即可实现自己的更高性能的序列化算法。Java序列化机制的速度比较慢，而且序列化后的数据占用的内存空间比较大。
Kryo序列化机制：Spark也支持使用Kryo类库来进行序列化。Kryo序列化机制比Java序列化机制更快，而且序列化后的数据占用的空间更小，通常比java序列化的数据占用的空间要小10倍、Kryo序列化机制之所以不是默认的序列化机智的原因是，有些类型虽然实现了Seriralizable接口，但是它也不一定能够进行序列化；此外，如果你要的到最佳的性能，Kryo还要求你在Spark应用程序汇总，对所有你需要序列化的类型都进行注册。

常规性能调优六：调节本地化等待时长

Spark 作业运行过程中，Driver 会对每一个 stage 的task 进行分配。根据 Spark 的task分配算法，Spark 希望 task 能够运行在它要计算的数据所在的节点(数据本地化思想),这样就可以避免数据的网络传输。通常来说,task 可能不会被分配到它处理的数据所在的节点，因为这些节点可用的资源可能已经用尽， 此时， Spark 会等待一段时间，默认 3s，如果等待指定时间后仍然无法在指定节点运行，那么会自动降级，尝试将 task 分配到比较差的本地化级别所对应的节点上，比如将 task 分配到离它要计算的数据比较近的一个节点，然后进行计算，如果当前级别仍然不行,那么继续降级。

当 task 要处理的数据不在 task 所在节点上时，会发生数据的传输。task 会通过所在节点的 BlockManager 获取数据， BlockManager 发现数据不在本地时，会通过网络传输组件从数据所在节点的 BlockManager 处获取数据。

网络传输数据的情况是我们不愿意看到的，大量的网络传输会严重影响性能， 因此，我们希望通过调节本地化等待时长， 如果在等待时长这段时间内， 目标节点处理完成了一部分 task，那么当前的 task 将有机会得到执行，这样就能够改善 Spark 作业的整体性能。

Spark 的本地化等级如表 2-3 所示：
表 2-3 Spark本地化等级

在 Spark 项目开发阶段，可以使用 client 模式对程序进行测试， 此时,可以在本地看到比较全的日志信息，日志信息中有明确的 task 数据本地化的级别，如果大部分都是 PROCESS_LOCAL，那么就无需进行调节，但是如果发现很多的级别都是NODE_LOCAL、ANY，那么需要对本地化的等待时长进行调节，通过延长本地化等待时长，看看 task 的本地化级别有没有提升，并观察 Spark 作业的运行时间有没有缩短。

注意,过犹不及， 不要将本地化等待时长延长地过长， 导致因为大量的等待时长，使得 Spark 作业的运行时间反而增加了。
Spark 本地化等待时长的设置如代码清单 2-5 所示：

代码清单  2-5 Spark本地化等待时长设置示例
 val conf = new SparkConf() .set("spark.locality.wait", "6")

资源调优

• 调优概述

在开发完Spark作业之后，就该为作业配置合适的资源了。Spark的资源参数，基本都可以在spark-submit命令中作为参数设置。很多Spark初学者，通常不知道该设置哪些必要的参数，以及如何设置这些参数，最后就只能胡乱设置，甚至压根儿不设置。资源参数设置的不合理，可能会导致没有充分利用集群资源，作业运行会极其缓慢；或者设置的资源过大，队列没有足够的资源来提供，进而导致各种异常。总之，无论是哪种情况，都会导致Spark作业的运行效率低下，甚至根本无法运行。因此我们必须对Spark作业的资源使用原理有一个清晰的认识，并知道在Spark作业运行过程中，有哪些资源参数是可以设置的，以及如何设置合适的参数值。

• Spark作业基本运行原理

image.png

详细原理见上图。我们使用spark-submit提交一个Spark作业之后，这个作业就会启动一个对应的Driver进程。根据你使用的部署模式（deploy-mode）不同，Driver进程可能在本地启动，也可能在集群中某个工作节点上启动。Driver进程本身会根据我们设置的参数，占有一定数量的内存和CPU core。而Driver进程要做的第一件事情，就是向集群管理器（可以是Spark Standalone集群，也可以是其他的资源管理集群，美团•大众点评使用的是YARN作为资源管理集群）申请运行Spark作业需要使用的资源，这里的资源指的就是Executor进程。YARN集群管理器会根据我们为Spark作业设置的资源参数，在各个工作节点上，启动一定数量的Executor进程，每个Executor进程都占有一定数量的内存和CPU core。
在申请到了作业执行所需的资源之后，Driver进程就会开始调度和执行我们编写的作业代码了。Driver进程会将我们编写的Spark作业代码分拆为多个stage，每个stage执行一部分代码片段，并为每个stage创建一批task，然后将这些task分配到各个Executor进程中执行。task是最小的计算单元，负责执行一模一样的计算逻辑（也就是我们自己编写的某个代码片段），只是每个task处理的数据不同而已。一个stage的所有task都执行完毕之后，会在各个节点本地的磁盘文件中写入计算中间结果，然后Driver就会调度运行下一个stage。下一个stage的task的输入数据就是上一个stage输出的中间结果。如此循环往复，直到将我们自己编写的代码逻辑全部执行完，并且计算完所有的数据，得到我们想要的结果为止。
Spark是根据shuffle类算子来进行stage的划分。如果我们的代码中执行了某个shuffle类算子（比如reduceByKey、join等），那么就会在该算子处，划分出一个stage界限来。可以大致理解为，shuffle算子执行之前的代码会被划分为一个stage，shuffle算子执行以及之后的代码会被划分为下一个stage。因此一个stage刚开始执行的时候，它的每个task可能都会从上一个stage的task所在的节点，去通过网络传输拉取需要自己处理的所有key，然后对拉取到的所有相同的key使用我们自己编写的算子函数执行聚合操作（比如reduceByKey()算子接收的函数）。这个过程就是shuffle。
当我们在代码中执行了cache/persist等持久化操作时，根据我们选择的持久化级别的不同，每个task计算出来的数据也会保存到Executor进程的内存或者所在节点的磁盘文件中。
因此Executor的内存主要分为三块：第一块是让task执行我们自己编写的代码时使用，默认是占Executor总内存的20%；第二块是让task通过shuffle过程拉取了上一个stage的task的输出后，进行聚合等操作时使用，默认也是占Executor总内存的20%；第三块是让RDD持久化时使用，默认占Executor总内存的60%。
task的执行速度是跟每个Executor进程的CPU core数量有直接关系的。一个CPU core同一时间只能执行一个线程。而每个Executor进程上分配到的多个task，都是以每个task一条线程的方式，多线程并发运行的。如果CPU core数量比较充足，而且分配到的task数量比较合理，那么通常来说，可以比较快速和高效地执行完这些task线程。
以上就是Spark作业的基本运行原理的说明，大家可以结合上图来理解。理解作业基本原理，是我们进行资源参数调优的基本前提。

• 资源参数调优

了解完了Spark作业运行的基本原理之后，对资源相关的参数就容易理解了。所谓的Spark资源参数调优，其实主要就是对Spark运行过程中各个使用资源的地方，通过调节各种参数，来优化资源使用的效率，从而提升Spark作业的执行性能。以下参数就是Spark中主要的资源参数，每个参数都对应着作业运行原理中的某个部分，我们同时也给出了一个调优的参考值。
num-executors

• 参数说明：该参数用于设置Spark作业总共要用多少个Executor进程来执行。Driver在向YARN集群管理器申请资源时，YARN集群管理器会尽可能按照你的设置来在集群的各个工作节点上，启动相应数量的Executor进程。这个参数非常之重要，如果不设置的话，默认只会给你启动少量的Executor进程，此时你的Spark作业的运行速度是非常慢的。
• 参数调优建议：每个Spark作业的运行一般设置50~100个左右的Executor进程比较合适，设置太少或太多的Executor进程都不好。设置的太少，无法充分利用集群资源；设置的太多的话，大部分队列可能无法给予充分的资源。
  executor-memory
• 参数说明：该参数用于设置每个Executor进程的内存。Executor内存的大小，很多时候直接决定了Spark作业的性能，而且跟常见的JVM OOM异常，也有直接的关联。
• 参数调优建议：每个Executor进程的内存设置4G_{8G较为合适。但是这只是一个参考值，具体的设置还是得根据不同部门的资源队列来定。可以看看自己团队的资源队列的最大内存限制是多少，num-executors乘以executor-memory，是不能超过队列的最大内存量的。此外，如果你是跟团队里其他人共享这个资源队列，那么申请的内存量最好不要超过资源队列最大总内存的1/3}1/2，避免你自己的Spark作业占用了队列所有的资源，导致别的同学的作业无法运行。
  executor-cores
• 参数说明：该参数用于设置每个Executor进程的CPU core数量。这个参数决定了每个Executor进程并行执行task线程的能力。因为每个CPU core同一时间只能执行一个task线程，因此每个Executor进程的CPU core数量越多，越能够快速地执行完分配给自己的所有task线程。
• 参数调优建议：Executor的CPU core数量设置为2~4个较为合适。同样得根据不同部门的资源队列来定，可以看看自己的资源队列的最大CPU core限制是多少，再依据设置的Executor数量，来决定每个Executor进程可以分配到几个CPU core。同样建议，如果是跟他人共享这个队列，那么num-executors * executor-cores不要超过队列总CPU core的1/3~1/2左右比较合适，也是避免影响其他同学的作业运行。
  driver-memory
• 参数说明：该参数用于设置Driver进程的内存。
• 参数调优建议：Driver的内存通常来说不设置，或者设置1G左右应该就够了。唯一需要注意的一点是，如果需要使用collect算子将RDD的数据全部拉取到Driver上进行处理，那么必须确保Driver的内存足够大，否则会出现OOM内存溢出的问题。
  spark.default.parallelism
• 参数说明：该参数用于设置每个stage的默认task数量。这个参数极为重要，如果不设置可能会直接影响你的Spark作业性能。
• 参数调优建议：Spark作业的默认task数量为500~1000个较为合适。很多同学常犯的一个错误就是不去设置这个参数，那么此时就会导致Spark自己根据底层HDFS的block数量来设置task的数量，默认是一个HDFS block对应一个task。通常来说，Spark默认设置的数量是偏少的（比如就几十个task），如果task数量偏少的话，就会导致你前面设置好的Executor的参数都前功尽弃。试想一下，无论你的Executor进程有多少个，内存和CPU有多大，但是task只有1个或者10个，那么90%的Executor进程可能根本就没有task执行，也就是白白浪费了资源！因此Spark官网建议的设置原则是，设置该参数为num-executors * executor-cores的2~3倍较为合适，比如Executor的总CPU core数量为300个，那么设置1000个task是可以的，此时可以充分地利用Spark集群的资源。
  spark.storage.memoryFraction
• 参数说明：该参数用于设置RDD持久化数据在Executor内存中能占的比例，默认是0.6。也就是说，默认Executor 60%的内存，可以用来保存持久化的RDD数据。根据你选择的不同的持久化策略，如果内存不够时，可能数据就不会持久化，或者数据会写入磁盘。
• 参数调优建议：如果Spark作业中，有较多的RDD持久化操作，该参数的值可以适当提高一些，保证持久化的数据能够容纳在内存中。避免内存不够缓存所有的数据，导致数据只能写入磁盘中，降低了性能。但是如果Spark作业中的shuffle类操作比较多，而持久化操作比较少，那么这个参数的值适当降低一些比较合适。此外，如果发现作业由于频繁的gc导致运行缓慢（通过spark web ui可以观察到作业的gc耗时），意味着task执行用户代码的内存不够用，那么同样建议调低这个参数的值。
  spark.shuffle.memoryFraction
• 参数说明：该参数用于设置shuffle过程中一个task拉取到上个stage的task的输出后，进行聚合操作时能够使用的Executor内存的比例，默认是0.2。也就是说，Executor默认只有20%的内存用来进行该操作。shuffle操作在进行聚合时，如果发现使用的内存超出了这个20%的限制，那么多余的数据就会溢写到磁盘文件中去，此时就会极大地降低性能。
• 参数调优建议：如果Spark作业中的RDD持久化操作较少，shuffle操作较多时，建议降低持久化操作的内存占比，提高shuffle操作的内存占比比例，避免shuffle过程中数据过多时内存不够用，必须溢写到磁盘上，降低了性能。此外，如果发现作业由于频繁的gc导致运行缓慢，意味着task执行用户代码的内存不够用，那么同样建议调低这个参数的值。
  资源参数的调优，没有一个固定的值，需要同学们根据自己的实际情况（包括Spark作业中的shuffle操作数量、RDD持久化操作数量以及spark web ui中显示的作业gc情况），同时参考本篇文章中给出的原理以及调优建议，合理地设置上述参数。

• 资源参数参考示例

以下是一份spark-submit命令的示例，大家可以参考一下，并根据自己的实际情况进行调节：

1./bin/spark-submit \ 
2--master yarn-cluster \ 
3--num-executors 100 \ 
4--executor-memory 6G \ 
5--executor-cores 4 \ 
6--driver-memory 1G \ 
7   --conf spark.default.parallelism=1000 \ 
8--conf spark.storage.memoryFraction=0.5 \ 
9   --conf spark.shuffle.memoryFraction=0.3 \