版权声明：本文为博主原创文章，转载请注明出处。

Spark调优秘诀

1.诊断内存的消耗

在Spark应用程序中，内存都消耗在哪了？

• 1.每个Java对象都有一个包含该对象元数据的对象头，其大小是16个Byte。由于在写代码时候，可能会出现这种情况：对象头比对象本身占有的字节数更多，比如对象只有一个int的域。一般这样设计是不合理的，造成了对象的“浪费”，在实际开发中应当避免这种情况。

• 2.Java的String对象，会比它内部的原始数据要多出40个字节。因为它内部使用char数组来保存内部的字符序列的，并且还得保存诸如数组长度之类的信息。而且String使用的是UTF-16编码，每个字符会占用2个字节。比如，包含10个字符的String，会占用60个字节。不过该使用String的时候就应该使用，这是无法避免的事，相对于后面说的序列化库、持久化、垃圾回收、提高并行度、广播共享数据、更有Shuffle阶段的优化等方面，String对象的内存特性就是毛毛雨了。

• 3.Java中的集合类型，比如HashMap和LinkedList，内部使用的是链式数据结构。既然是链表，那么每个节点都有额外的信息来保证前后节点的查寻，它们都使用了Entry对象来包装。Entry对象不仅仅有对象头，还有指向下一个Entry的指针，通常占用8个字节。

• 4.元素类型为原始数据类型（比如int）的集合，内部通常会使用原始数据类型的包装类型，比如Integer，来存储元素。这种情况其实和第三种情况一致的，都是因为Java的自动装箱和拆箱而导致的。

以上就是Spark应用程序针对开发语言的特性所占用的内存大小，要通过什么办法来查看和确定消耗内存大小呢？

• 1、自行设置RDD的并行度。有两种方式：第一，在parallelize()、textFile()等外部数据源方法中传入第二个参数，设置RDD的task / partition的数量；第二，用SparkConf.set()方法，设置参数(spark.default.parallelism)，统一设置整个Spark Application所有RDD的partition数量。

• 2、调用RDD.cache()将RDD cache到内存中。方便直接从log信息中看出每个partition消耗的内存。

• 3、找出Driver进程打印的log信息，会有类似于：“INFO BlockManagerMasterActor: Added rdd01 in memory on mbk.local:50311 (size: 717.5 KB, free: 555.5 MB)”的日志信息。这就显示了每个partition占用了多少内存。

• 4、将这个内存信息乘以partition数量，即可得出RDD的内存占用量。

注意，这种方法，一定要确保电脑的内存能够承受测试的数据，不然会报出oom异常。

通过以上的简介，大概知道了内存的消耗和如何查看消耗的内存了。但是只知道内存的消耗而不去优化它，肯定是不行的，在生产环境中，每一分每一秒都是金钱和客户的满意。比如这个报表要求每天早上8点跑出结果给领导看的，然而因为你的Spark程序实在太慢了，11点才出结果，那么领导显然会不满意的，最后奖金就变少了。因此下面来根据多个方面来逐点分析如何对Spark应用程序调优，分析的顺序是从表面到底层的Shuffle阶段。其实最重要的调优还是Shuffle阶段的调优。

2.高性能序列化类库

在分布式应用程序中，要想程序能够工作，首先第一步是什么？毫无疑问是分布式节点之间的通信，要想通信，最重要的阶段是序列化和反序列化。那么，显而易见，速度更快，更稳定的序列化库影响分布式系统的通信效率。

在Spark中，默认是使用Java自带的序列化机制——基于ObjectInputStream和ObjectOutputStream的序列化机制，这是为了提高便捷性和适用性，毕竟是Java原生的嘛。然鹅，自带的东西往往考虑的东西比较多，没法做到样样俱全，比如内序列化后占据的内存还是较大，但是Spark是基于内存的大数据框架，对内存的要求很高。所以，在Spark应用程序中，Java自带的序列化库的效率有点差强人意。需求是从实际出发的嘛，最终Spark也提供了另外一种序列化机制——Kryo序列化机制。

Kryo序列化机制比Java序列化机制更快，序列化后的数据占的内存更小。那么Kryo序列化机制这么好，为什么不选用它是默认序列化库呢？这里提一句话“人无完人，谁能无错”，Kryo序列化机制也样，之所以不选用它为默认序列化机制是因为有些类型虽然实现了Seriralizable接口，但是不一定能够进行序列化；此外，如果要得到最佳的性能，需要在Spark应用程序中，对所有 需要序列化的类型都进行注册。

使用Kryo序列化机制的方法： 1.给SparkConf加入一个参数 SparkConf().set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

2.对需要序列化的类自行进行注册(因为如果不注册，Kryo必须一直保存类型的全限定名，会占用内存。Spark默认是对Scala中常用的类型自动注册了Kryo的，都在AllScalaRegistry类中) Scala版本：

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.registerKryoClasses(Array(classOf[Counter] ))
val sc = new SparkContext(conf)

Java版本：

SparkConf conf = new SparkConf().setMaster(...).setAppName(...)
conf.registerKryoClasses(Counter.class)
JavaSparkContext sc = new JavaSparkContext(conf)

还可以对Kryo序列化机制进行优化达到更优的效果。

• 1、优化缓存大小。如果注册的要序列化的自定义的类型，本身很大大，比如包含了超过100个field。会导致要序列化的对象过大。此时需要对Kryo本身进行优化。因为Kryo内部的缓存可能不够存放这么大的class对象。此时需要调用SparkConf.set()方法，设置spark.kryoserializer.buffer.mb参数的值，将其调大以适用。默认情况下spark.kryoserializer.buffer.mb是2，即最大能缓存2M的对象，然后进行序列化。可以在必要时将其调大。比如设置为10。

• 2、预先注册自定义类型。虽然不注册自定义类型，Kryo类库也能正常工作，但是那样对于它要序列化的每个对象，都会保存一份它的全限定类名。反而会耗费大量内存。因此通常都预先注册好要序列化的自定义的类。

总结，需要用到Kryo序列化机制的场景，算子内部使用了外部的大对象或者大数据结构。那么可以切换到Kryo序列化，序列化速度更快，和获得更小的序列化数据，减少内存的消耗。

3.优化数据结构

对数据结构的优化，主要是针对Java数据结构(如果用scala开发的话，其实原理也一样的)。其实就是算子里面的局部变量或者算子函数外部的数据结构。比如基于链式结构的数据结构、包装类型的数据结构等，它们在除了本身的数据之外，还会有额外的数据信息来维持它们的数据类型，这样就会比预想占有更大的内存。

以下是一些优化建议：

• 1、能使用数组或字符串就不要用集合类。即优先使用Array，退而求次才是ArrayList、LinkedList、HashMap、HashTable等。熟悉Java语言的都知道集合类一般是泛型的，然鹅泛型的类型是包装类，比如List list = new ArrayList()，就会因为包装类而占有额外的内存，最后占有更多的额外开销。在生产开发中的做法是，对于HashMap、List这种数据，统一用String拼接成特殊格式的字符串。如Map<Integer, Person> persons = new HashMap<Integer, Person>()。可以优化为，特殊的字符串格式：id:name,address|id:name,address...

• 2、避免使用多层嵌套的对象结构。public class Teacher { private List students = new ArrayList() }。就是非常不好的例子。因为Teacher类的内部又嵌套了大量的小Student对象。比如说，对于上述例子，也完全可以使用特殊的字符串来进行数据的存储。比如，用json字符串来存储数据，就是一个很好的选择。{"teacherId": 1, "teacherName": "leo", students:[{"studentId": 1, "studentName": "tom"},{"studentId":2, "studentName":"marry"}]}

• 3、能用int就不用String。虽然String比集合咧更高效，但是之前说过Java的String是占2个字节的，使用int会优化内存。

总结，在写Spark程序的时候，要牢牢记住，尽量压榨因语言带来的内存开销，达到节约内存的目的。

4.对多次使用的RDD进行持久化或Checkpoint

• 1、对一个RDD，基于它进行了多次transformation或者action操作。非常有必要对其进行持久化操作，以避免对一个RDD反复进行计算。

• 2、如果要保证在RDD的持久化数据可能丢失的情况下，还要保证高性能，那么可以对RDD进行Checkpoint操作。

如下图，两次对同一个RDD操作，但是比如当路径1先计算完RDD2得出RDD3，当RDD5再次计算RDD2的时候，由于在Spark中，对RDD计算后，如果没有持久化，在计算后可能就会立刻抛弃掉数据。所以第二次计算RDD2时需要重新计算RDD2前面的RDD。这样很明显就消耗了额外的时间。

总结，对于后面要多次可能用到的RDD，要对其持久化，如果要高可用，更要对其checkpoint，保证以后出错节省大量的时间。正所谓“长痛不如短痛”，一时的付出是为了后面的快速恢复错误和高可用。

5.使用序列化的持久化级别

• RDD的数据是持久化到内存，或者磁盘中的。但是如果机器的内存大小不是很充足，或者有时为了节省机器的内存开销，比如在生产环境下，机器不单单是跑这么一个Spark应用的,还需要留些内存供其他应用使用。这种情况下，可以使用序列化的持久化级别。比如MEMORYONLYSER、MEMORYANDDISKSER等。用法是：RDD.persist(StorageLevel.MEMORYONLY_SER)。

• 将数据序列化之后，再持久化，可以大大减小对内存的消耗。此外，数据量小了之后，如果要写入磁盘，磁盘io性能消耗也比较小。

• 对RDD持久化序列化后，RDD的每个partition的数据，都是序列化为一个巨大的字节数组。这样，对于内存的消耗就小了。但是唯一的缺点是获取RDD数据时，需要对其进行反序列化，会增大其性能开销。这种情况下可以使用第二点的Kryo序列化机制配合，提高序列化的效率。

6.Java虚拟机垃圾回收调优

7.提高并行度

• 在实际使用Spark集群的时候，很多时候对于集群的资源并不是一定会被充分利用到，这是由于task和cpu核的协调不好导致的。要想合理的“榨干”集群的资源和性能，可以合理的设置Spark应用程序运行的并行度，来充分地利用集群的资源，这样才能充分的提高Spark应用程序的性能。

• Spark的数据源有两种，一种是外部的，比如HDFS等分布式文件系统，或者通过现有的数组等数据结构序列化而成；一种是通过已有的RDD转换而来的。这里以Spark读取HDFS的数据为例子。Spark会根据读取HDFS的时候把每个block划分为一个Partition，其实也是按照这个来自动设置并行度的。对于reduceByKey等会发生shuffle的算子操作，会使用并行度最大的父RDD的并行度作为Spark应用的并行度。

• 通过上面的分析，我们可以手动设置并行度，在读取HDFS或者并行化数据的时候调用textFile()和parallelize()等方法的时候，通过第二个参数来设置并行度。也可以使用spark.default.parallelism参数，来设置统一的并行度。根据Spark官方的推荐，最优的方案是给集群中的每个cpu core设置2~3个task，也就是task的数量是cpu核的2~3倍。

• 以下是实现例子：现在已知cpu core有10个。比如spark-submit设置了executor数量是2个，每个executor有5个core。但是在Spark应用程序中这样设置了SparkConf().set("spark.default.parallelism", "5")，那么application总共会有5个core。实际上所有的RDD都被设为了partition为5，也就是每个RDD的数据分为5份，也就是5份数据(partition)成为5个task分配到这两个executor中。很明显，Spark应用程序在运行的时候，只占用了5个cpu core，还剩下5个cpu core是没用到的，浪费了集群资源。此时可以设置这样来优化Spark的集群性能，通过设置参数 SparkConf().set("spark.default.parallelism", "30")来设置合理的并行度，从而充分利用资源。为什么呢？请看下图：

8.广播共享数据

• RDD实质是弹性分布式数据集，在每个节点中的每个task(一个节点可以有很多个task)操作的只是RDD的一部分数据，如果RDD算子操作使用到了算子函数外部的一份大数据的时候，实际上是Spark应用程序把数据文件通过driver发送给每一个节点的每一个task，很明显，这样会造成大量的网络IO操作，大量消耗节点上的内存。其实很容易想到，把一份大数据文件发送给每个节点就OK了，单个节点的所有task共享一份数据，这样就会节省大量的网络IO操作和节省大量内存消耗。

• 如果算子函数中，使用到了特别大的数据(比如一份大的配置文件)供每个节点的所有task使用，可以借助Spark提供的共享变量。共享变量有两种，一是广播变量，一是累加器。广播变量是只读的，通常用来提供一份数据给所有的节点，每个节点的task访问访问同一份数据。而累加器是可写可读的，一个累加器一般是用于所有节点对用一个简单的整型变量进行共享累加，共同维护一份数据。这样的话，就不至于将一个大数据拷贝到每一个task上去。而是给每个节点拷贝一份，然后节点上的task共享该数据。原理如图所示： 

9.数据本地化

Spark数据本地化的基本原理

• Spark和MapReduce是如今两个最流行的大数据框架，它们的原理都是计算移动，而数据不移动，计算找数据。这样做的创新性是避免了大量数据的网络传输造成网络IO和内存的消耗。因此引出一个叫“数据本地化”的概念。

• 数据本地化对于Spark Job性能有着巨大的影响。如果数据以及要计算它的代码是在同一个节点，性能会非常高。但是，如果数据和计算它的代码是位于不同的节点，那么其中之一必须到另外一方的机器上。通常来说，移动代码到其他节点，会比移动数据到代码所在的节点上去，速度要快得多，因为代码比较小。Spark也正是基于这个数据本地化的原则来构建task调度算法的。

• 数据本地化，指的是，数据离计算它的代码有多近。基于数据距离代码的距离，有几种数据本地化级别：

  • 1、PROCESS_LOCAL：数据和计算它的代码在同一个JVM进程中。
  • 2、NODE_LOCAL：数据和计算它的代码在一个节点上，但是不在一个进程中，比如在不同的executor进程中，或者是数据在HDFS文件的block中。
  • 3、NO_PREF：数据从哪里过来，性能都是一样的。
  • 4、RACK_LOCAL：数据和计算它的代码在一个机架上。
  • 5、ANY：数据可能在任意地方，比如其他网络环境内，或者其他机架上。

Spark数据本地化的特点

• Spark倾向于使用最好的本地化级别来调度task，但并不是每次都会使用最好的本地化数据的。在实际中，如果没有任何未处理的数据在空闲的executor上，Spark会放低本地化级别。这时有两个选择：第一，driver等待，直到executor上的cpu释放出来，就分配task等资源给这个executor；第二，立即在任意一个executor上启动一个task。

• Spark会默认等待一段时间(这个事件可以通过参数来设置)，来期望在task要处理的数据所在的节点上的executor空闲出一个cpu，从而为其分配task鞥资源。但只要超过了时间，Spark就会将task分配到其他任意一个空闲的executor上。

• 可以设置参数，spark.locality系列参数，来调节Spark等待task可以进行数据本地化的时间。spark.locality.wait（3000毫秒）、spark.locality.wait.node、spark.locality.wait.process、spark.locality.wait.rack

• 针对以上的分析，我们可以这样调优，增大查找本地化数据的超时时间和重试次数，因为时间更长更利于查找本地化数据的节点的executor，重试次数越多，更多机会尝试查找本地化数据的节点的executor。

• 调优方式，主要是spark.locality.wait（3000毫秒）、spark.locality.wait.node、spark.locality.wait.process、spark.locality.wait.rack这些参数，具体的根据实际的业务需求来控制参数就OK了。

10.reduceByKey和groupByKey的选择

• 以下两种方式是等价的，但是实现的原理却不相同。reduceByKey，因为它会在map端，先进行本地combine，可以大大减少要传输到reduce端的数据量，减小网络传输的开销。而groupByKey算子却不会这样优化。所以只有在reduceByKey处理不了时，才用groupByKey().map()来替代。

  val counts = pairs.reduceByKey(_ + _)
  val counts = pairs.groupByKey().map(wordCounts => (wordCounts._1, wordCounts._2.sum))
  

11.shuffle性能优化

• 无论是MapReduce还是Spark，Shuffle阶段是最重要的阶段，它的好坏影响着整个Spark的性能。其实Shuffle阶段的调优，可以从以下的参数入手：
  • new SparkConf().set("spark.shuffle.consolidateFiles", "true")
  • spark.shuffle.consolidateFiles：是否开启shuffle block file的合并，默认为false
  • spark.reducer.maxSizeInFlight：reduce task的拉取缓存，默认48m
  • spark.shuffle.file.buffer：map task的写磁盘缓存，默认32k
  • spark.shuffle.io.maxRetries：拉取失败的最大重试次数，默认3次
  • spark.shuffle.io.retryWait：拉取失败的重试间隔，默认5s
  • spark.shuffle.memoryFraction：用于reduce端聚合的内存比例，默认0.2，超过比例就会溢出到磁盘上

原理请先看下图，然后再做分析。 

这个是没有开启consolidateFiles优化(Spark1.3之后加入的)，会产生大量的磁盘文件，在写磁盘和result task拉取数据的时候，会浪费过多的系统资源。

开启consolidateFiles优化 

优化方法：

• 开启consolidateFiles，增大result task的拉取缓存，增大shufflemaptask的写磁盘缓存，增大重试次数和重试间隔，调大用于reduce端聚合的内存比例
• new SparkConf().set("spark.shuffle.consolidateFiles", "true")
• spark.shuffle.consolidateFiles：是否开启shuffle block file的合并，默认为false
• spark.reducer.maxSizeInFlight：ResultTask的拉取缓存，默认48m
• spark.shuffle.file.buffer：map task的写磁盘缓存，默认32k
• spark.shuffle.io.maxRetries：拉取失败的最大重试次数，默认3次
• spark.shuffle.io.retryWait：拉取失败的重试间隔，默认5s

• spark.shuffle.memoryFraction：用于reduce端聚合的内存比例，默认0.2，超过比例就会溢出到磁盘上

• 影响一个Spark作业性能的因素，主要还是代码开发、资源参数以及数据倾斜，shuffle调优只能在整个Spark的性能调优中占到一小部分。

• shuffle read的拉取过程是一边拉取一边进行聚合的。每个shuffle read task都会有一个自己的buffer缓冲，每次都只能拉取与buffer缓冲(这个缓存大小可以通过上面的参数来设定)相同大小的数据，然后通过内存中的一个Map进行聚合等操作。聚合完一批数据后，再拉取下一批数据，并放到buffer缓冲中进行聚合操作。一直循环，直到最后将所有数据到拉取完，并得到最终的结果。

• 开启consolidate机制之后，在shuffle write过程中，task就不是为下游stage的每个task创建一个磁盘文件了。此时会出现shuffleFileGroup的概念，每个shuffleFileGroup会对应一批磁盘文件，磁盘文件的数量与下游stage的task数量是相同的。一个Executor上有多少个CPU core，就可以并行执行多少个task。而第一批并行执行的每个task都会创建一个shuffleFileGroup，并将数据写入对应的磁盘文件内。

• 当Executor的CPU core执行完一批task，接着执行下一批task时，下一批task就会复用之前已有的shuffleFileGroup，包括其中的磁盘文件。也就是说，此时task会将数据写入已有的磁盘文件中，而不会写入新的磁盘文件中。因此，consolidate机制允许不同的task复用同一批磁盘文件，这样就可以有效将多个task的磁盘文件进行一定程度上的合并，从而大幅度减少磁盘文件的数量，进而提升shuffle write的性能。

Spark的调优至此告一段落，下一篇会针对实际过程中遇到的问题特定的调优。