Spark性能优化技术

  1. Spark的性能优化，主要手段包括：
  2. 1、使用高性能序列化类库
  3. 2、优化数据结构
  4. 3、对多次使用的RDD进行持久化 / Checkpoint
  5. 4、使用序列化的持久化级别
  6. 5、Java虚拟机垃圾回收调优
  7. 6、提高并行度
  8. 7、广播共享数据
  9. 8、数据本地化
  10. 9、reduceByKey和groupByKey的合理使用
  11. 10、Shuffle调优（核心中的核心，重中之重）

诊断内存的消耗

内存都花费在哪里了？

  1. 1、每个Java对象，都有一个对象头，会占用16个字节，主要是包括了一些对象的元信息，比如指向它的类的指针。如果一个对象本身很小，比如就包括了一个int类型的field，那么它的对象头实际上比对象自己还要大。
  3. 2、Java的String对象，会比它内部的原始数据，要多出40个字节。因为它内部使用char数组来保存内部的字符序列的，并且还得保存诸如数组长度之类的信息。而且因为String使用的是UTF-16编码，所以每个字符会占用2个字节。比如，包含10个字符的String，会占用60个字节。
  5. 3、Java中的集合类型，比如HashMap和LinkedList，内部使用的是链表数据结构，所以对链表中的每一个数据，都使用了Entry对象来包装。Entry对象不光有对象头，还有指向下一个Entry的指针，通常占用8个字节。
  7. 4、元素类型为原始数据类型（比如int）的集合，内部通常会使用原始数据类型的包装类型，比如Integer，来存储元素。

如何判断你的程序消耗了多少内存？

  1. 这里有一个非常简单的办法来判断，你的spark程序消耗了多少内存。
  3. 1、首先，自己设置RDD的并行度，有两种方式：要不然，在parallelize()、textFile()等方法中，传入第二个参数，设置RDD的task / partition的数量；要不然，用SparkConf.set()方法，设置一个参数，spark.default.parallelism，可以统一设置这个application所有RDD的partition数量。
  5. 2、其次，在程序中将RDD cache到内存中，调用RDD.cache()方法即可。
  7. 3、最后，观察Driver的log，你会发现类似于：“INFO BlockManagerMasterActor: Added rdd_0_1 in memory on mbk.local:50311 (size: 717.5 KB, free: 332.3 MB)”的日志信息。这就显示了每个partition占用了多少内存。
  9. 4、将这个内存信息乘以partition数量，即可得出RDD的内存占用量。

高性能序列化类库

Spark提供的两种序列化机制

  1. 1、Java序列化机制：默认情况下，Spark使用Java自身的ObjectInputStream和ObjectOutputStream机制进行对象的序列化
  3. 2、Kryo序列化机制：Spark也支持使用Kryo类库来进行序列化。
  4.     Kryo序列化机制比Java序列化机制更快，而且序列化后的数据占用的空间更小，通常比Java序列化的数据占用的空间要小10倍。

如何使用Kryo序列化机制（一）

  1. 首先要用SparkConf设置一个参数，使用new SparkConf().set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")即可，即将Spark的序列化器设置为KryoSerializer。

如何使用Kryo序列化机制（二）

  1. 如果要注册自定义的类型，那么就使用如下的代码，即可：
  3. Scala版本：
  4. val conf = new SparkConf().setMaster(...).setAppName(...)
  5. conf.registerKryoClasses(Array(classOf[Counter] ))
  6. val sc = new SparkContext(conf)
  8. Java版本：
  9. SparkConf conf = new SparkConf().setMaster(...).setAppName(...)
  10. conf.registerKryoClasses(Counter.class)
  11. JavaSparkContext sc = new JavaSparkContext(conf)

优化Kryo类库的使用

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

实例

  1. import com.esotericsoftware.kryo.Kryo
  2. import org.apache.spark.serializer.KryoRegistrator
  4. class MyRegistrator extends KryoRegistrator {
  5.   override def registerClasses(kryo: Kryo) {
  6.     kryo.register(classOf[MyClass1])
  7.     kryo.register(classOf[MyClass2])
  8.   }
  9. }
  11. val conf = new SparkConf().setMaster(...).setAppName(...)
  12. conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  13. conf.set("spark.kryo.registrator", "mypackage.MyRegistrator")
  14. val sc = new SparkContext(conf)

在什么场景下使用Kryo序列化类库？

  1. 1.算子函数使用到了外部的大数据的情况
  2.     比如说吧，我们在外部定义了一个封装了应用所有配置的对象，比如自定义了一个MyConfiguration对象，里面包含了100m的数据。然后，在算子函数里面，使用到了这个外部的大对象

优化数据结构

优化数据结构

  1. 优化什么数据结构？其实主要就是优化你的算子函数，内部使用到的局部数据，或者是算子函数外部的数据。都可以进行数据结构的优化。优化之后，都会减少其对内存的消耗和占用。

如何优化数据结构（一）

  1. 1、优先使用数组以及字符串，而不是集合类。也就是说，优先用array，而不是ArrayList、LinkedList、HashMap等集合。
  3.     比如，有个List<Integer> list = new ArrayList<Integer>()，将其替换为int[] arr = new int[]。这样的话，array既比List少了额外信息的存储开销，还能使用原始数据类型（int）来存储数据，比List中用Integer这种包装类型存储数据，要节省内存的多。
  5. 还比如，通常企业级应用中的做法是，对于HashMap、List这种数据，统一用String拼接成特殊格式的字符串，比如Map<Integer, Person> persons = new HashMap<Integer, Person>()。可以优化为，特殊的字符串格式：id:name,address|id:name,address...。

如何优化数据结构（二）

  1. 2、避免使用多层嵌套的对象结构。比如说，public class Teacher { private List<Student> students = new ArrayList<Student>() }。就是非常不好的例子。因为Teacher类的内部又嵌套了大量的小Student对象。
  3. 比如说，对于上述例子，也完全可以使用特殊的字符串来进行数据的存储。比如，用json字符串来存储数据，就是一个很好的选择。
  5. {"teacherId": 1, "teacherName": "leo", students:[{"studentId": 1, "studentName": "tom"},{"studentId":2, "studentName":"marry"}]}
  7. 3、对于有些能够避免的场景，尽量使用int替代String。因为String虽然比ArrayList、HashMap等数据结构高效多了，占用内存量少多了，但是之前分析过，还是有额外信息的消耗。比如之前用String表示id，那么现在完全可以用数字类型的int，来进行替代。
  9. 这里提醒，在spark应用中，id就不要用常用的uuid了，因为无法转成int，就用自增的int类型的id即可。（sdfsdfdf-234242342-sdfsfsfdfd）

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

  1. 如果程序中，对某一个RDD，基于它进行了多次transformation或者action操作。那么就非常有必要对其进行持久化操作，以避免对一个RDD反复进行计算。
  3. 此外，如果要保证在RDD的持久化数据可能丢失的情况下，还要保证高性能，那么可以对RDD进行Checkpoint操作。

使用序列化的持久化级别

  1. 除了对多次使用的RDD进行持久化操作之外，还可以进一步优化其性能。因为很有可能，RDD的数据是持久化到内存，或者磁盘中的。那么，此时，如果内存大小不是特别充足，完全可以使用序列化的持久化级别，比如MEMORY_ONLY_SER、MEMORY_AND_DISK_SER等。使用RDD.persist(StorageLevel.MEMORY_ONLY_SER)这样的语法即可。
  3. 这样的话，将数据序列化之后，再持久化，可以大大减小对内存的消耗。此外，数据量小了之后，如果要写入磁盘，那么磁盘io性能消耗也比较小。
  5. 对RDD持久化序列化后，RDD的每个partition的数据，都是序列化为一个巨大的字节数组。这样，对于内存的消耗就小的多了。但是唯一的缺点就是，获取RDD数据时，需要对其进行反序列化，会增大其性能开销。
  7. 因此，对于序列化的持久化级别，还可以进一步优化，也就是说，使用Kryo序列化类库，这样，可以获得更快的序列化速度，并且占用更小的内存空间。但是要记住，如果RDD的元素（RDD<T>的泛型类型），是自定义类型的话，在Kryo中提前注册自定义类型。

Java虚拟机垃圾回收调优

Java虚拟机垃圾回收调优的背景

  1. 如果在持久化RDD的时候，持久化了大量的数据，那么Java虚拟机的垃圾回收就可能成为一个性能瓶颈。因为Java虚拟机会定期进行垃圾回收，此时就会追踪所有的java对象，并且在垃圾回收时，找到那些已经不在使用的对象，然后清理旧的对象，来给新的对象腾出内存空间。
  3. 垃圾回收的性能开销，是跟内存中的对象的数量，成正比的。所以，对于垃圾回收的性能问题，首先要做的就是，使用更高效的数据结构，比如array和string；其次就是在持久化rdd时，使用序列化的持久化级别，而且用Kryo序列化类库，这样，每个partition就只是一个对象——一个字节数组。

监测垃圾回收

  1. 我们可以对垃圾回收进行监测，包括多久进行一次垃圾回收，以及每次垃圾回收耗费的时间。只要在spark-submit脚本中，增加一个配置即可，--conf "spark.executor.extraJavaOptions=-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps"。
  3. 但是要记住，这里虽然会打印出Java虚拟机的垃圾回收的相关信息，但是是输出到了worker上的日志中，而不是driver的日志中。
  5. 但是这种方式也只是一种，其实也完全可以通过SparkUI（4040端口）来观察每个stage的垃圾回收的情况。

优化executor内存比例

  1. 对于垃圾回收来说，最重要的就是调节RDD缓存占用的内存空间，与算子执行时创建的对象占用的内存空间的比例。默认情况下，Spark使用每个executor 60%的内存空间来缓存RDD，那么在task执行期间创建的对象，只有40%的内存空间来存放。
  3. 在这种情况下，很有可能因为你的内存空间的不足，task创建的对象过大，那么一旦发现40%的内存空间不够用了，就会触发Java虚拟机的垃圾回收操作。因此在极端情况下，垃圾回收操作可能会被频繁触发。
  5. 在上述情况下，如果发现垃圾回收频繁发生。那么就需要对那个比例进行调优，使用new SparkConf().set("spark.storage.memoryFraction", "0.5")即可，可以将RDD缓存占用空间的比例降低，从而给更多的空间让task创建的对象进行使用。
  7. 因此，对于RDD持久化，完全可以使用Kryo序列化，加上降低其executor内存占比的方式，来减少其内存消耗。给task提供更多的内存，从而避免task的执行频繁触发垃圾回收。

高级垃圾回收调优（一）

  1. Java堆空间被划分成了两块空间，一个是年轻代，一个是老年代。年轻代放的是短时间存活的对象，老年代放的是长时间存活的对象。年轻代又被划分了三块空间，Eden、Survivor1、Survivor2。
  3. 首先，Eden区域和Survivor1区域用于存放对象，Survivor2区域备用。创建的对象，首先放入Eden区域和Survivor1区域，如果Eden区域满了，那么就会触发一次Minor GC，进行年轻代的垃圾回收。Eden和Survivor1区域中存活的对象，会被移动到Survivor2区域中。然后Survivor1和Survivor2的角色调换，Survivor1变成了备用。
  5. 如果一个对象，在年轻代中，撑过了多次垃圾回收，都没有被回收掉，那么会被认为是长时间存活的，此时就会被移入老年代。此外，如果在将Eden和Survivor1中的存活对象，尝试放入Survivor2中时，发现Survivor2放满了，那么会直接放入老年代。此时就出现了，短时间存活的对象，进入老年代的问题。
  7. 如果老年代的空间满了，那么就会触发Full GC，进行老年代的垃圾回收操作。

高级垃圾回收调优（二）

  1. Spark中，垃圾回收调优的目标就是，只有真正长时间存活的对象，才能进入老年代，短时间存活的对象，只能呆在年轻代。不能因为某个Survivor区域空间不够，在Minor GC时，就进入了老年代。从而造成短时间存活的对象，长期呆在老年代中占据了空间，而且Full GC时要回收大量的短时间存活的对象，导致Full GC速度缓慢。
  3. 如果发现，在task执行期间，大量full gc发生了，那么说明，年轻代的Eden区域，给的空间不够大。此时可以执行一些操作来优化垃圾回收行为：
  4. 1、包括降低spark.storage.memoryFraction的比例，给年轻代更多的空间，来存放短时间存活的对象；
  5. 2、给Eden区域分配更大的空间，使用-Xmn即可，通常建议给Eden区域，预计大小的4/3；
  6. 3、如果使用的是HDFS文件，那么很好估计Eden区域大小，如果每个executor有4个task，然后每个hdfs压缩块解压缩后大小是3倍，此外每个hdfs块的大小是64M，那么Eden区域的预计大小就是：4 * 3 * 64MB，然后呢，再通过-Xmn参数，将Eden区域大小设置为4 * 3 * 64 * 4/3。

最后一点总结

  1. 其实啊，根据经验来看，对于垃圾回收的调优，尽量就是说，调节executor内存的比例就可以了。因为jvm的调优是非常复杂和敏感的。除非是，真的到了万不得已的地方，然后呢，自己本身又对jvm相关的技术很了解，那么此时进行eden区域的调节，调优，是可以的。
  3. 一些高级的参数
  4. -XX:SurvivorRatio=4：如果值为4，那么就是两个Survivor跟Eden的比例是2:4，也就是说每个Survivor占据的年轻代的比例是1/6，所以，你其实也可以尝试调大Survivor区域的大小。
  5. -XX:NewRatio=4：调节新生代和老年代的比例

提高并行度

提高并行度

  1. 实际上Spark集群的资源并不一定会被充分利用到，所以要尽量设置合理的并行度，来充分地利用集群的资源。才能充分提高Spark应用程序的性能。
  3. Spark会自动设置以文件作为输入源的RDD的并行度，依据其大小，比如HDFS，就会给每一个block创建一个partition，也依据这个设置并行度。对于reduceByKey等会发生shuffle的操作，就使用并行度最大的父RDD的并行度即可。
  5. 可以手动使用textFile()、parallelize()等方法的第二个参数来设置并行度；也可以使用spark.default.parallelism参数，来设置统一的并行度。Spark官方的推荐是，给集群中的每个cpu core设置2~3个task。
  7. 比如说，spark-submit设置了executor数量是10个，每个executor要求分配2个core，那么application总共会有20个core。此时可以设置new SparkConf().set("spark.default.parallelism", "60")来设置合理的并行度，从而充分利用资源。

广播共享数据

广播共享数据

  1. 如果你的算子函数中，使用到了特别大的数据，那么，这个时候，推荐将该数据进行广播。这样的话，就不至于将一个大数据拷贝到每一个task上去。而是给每个节点拷贝一份，然后节点上的task共享该数据。
  3. 这样的话，就可以减少大数据在节点上的内存消耗。并且可以减少数据到节点的网络传输消耗。

数据本地化

数据本地化背景

  1. 数据本地化对于Spark Job性能有着巨大的影响。如果数据以及要计算它的代码是在一起的，那么性能当然会非常高。但是，如果数据和计算它的代码是分开的，那么其中之一必须到另外一方的机器上。通常来说，移动代码到其他节点，会比移动数据到代码所在的节点上去，速度要快得多，因为代码比较小。Spark也正是基于这个数据本地化的原则来构建task调度算法的。
  3. 数据本地化，指的是，数据离计算它的代码有多近。基于数据距离代码的距离，有几种数据本地化级别：
  4. 1、PROCESS_LOCAL：数据和计算它的代码在同一个JVM进程中。
  5. 2、NODE_LOCAL：数据和计算它的代码在一个节点上，但是不在一个进程中，比如在不同的executor进程中，或者是数据在HDFS文件的block中。
  6. 3、NO_PREF：数据从哪里过来，性能都是一样的。
  7. 4、RACK_LOCAL：数据和计算它的代码在一个机架上。
  8. 5、ANY：数据可能在任意地方，比如其他网络环境内，或者其他机架上。

数据本地化优化

  1. Spark倾向于使用最好的本地化级别来调度task，但是这是不可能的。如果没有任何未处理的数据在空闲的executor上，那么Spark就会放低本地化级别。这时有两个选择：第一，等待，直到executor上的cpu释放出来，那么就分配task过去；第二，立即在任意一个executor上启动一个task。
  3. Spark默认会等待一会儿，来期望task要处理的数据所在的节点上的executor空闲出一个cpu，从而将task分配过去。只要超过了时间，那么Spark就会将task分配到其他任意一个空闲的executor上。
  5. 可以设置参数，spark.locality系列参数，来调节Spark等待task可以进行数据本地化的时间。spark.locality.wait（3000毫秒）、spark.locality.wait.node、spark.locality.wait.process、spark.locality.wait.rack。

reduceByKey 和 groupByKey

  1. val counts = pairs.reduceByKey(_ + _)
  3. val counts = pairs.groupByKey().map(wordCounts => (wordCounts._1, wordCounts._2.sum))
  5. 如果能用reduceByKey，那就用reduceByKey，因为它会在map端，先进行本地combine，可以大大减少要传输到reduce端的数据量，减小网络传输的开销。
  7. 只有在reduceByKey处理不了时，才用groupByKey().map()来替代

groupByKey原理 和 reduceByKey 原理

  1. reduceByKey
  2.     对key对应的values进行聚合为一个值的场景，那么用reduceByKey是非常合适的，因为会先在ShuffleMapTask端写入本地磁盘文件的时候，进行本地聚合，再写入磁盘文件，此时，就会导致数据量大幅度缩减，甚至可能达到数据量缩减了几倍，甚至几十倍
  4. groupByKey原理
  5.     总结：groupByKey的性能，相对来说，是有问题的
  6. 因为，它是不会进行本地聚合的，而是原封不动的，把ShuffleMapTask的输出，拉取到ResultTask的内存中，从而导致网络传输的性能开销非常大
  8. 话虽如此，有些场景下，用其他算法，实现不了的，比如reduceByKey，
  9. sortByKey,countByKey实现不了的话，还是只能用groupByKey().map()来实现，可能你需要拿到某个key对应的所有的value，进行自定义的业务逻辑处理

shuffle性能优化

  1. new SparkConf().set("spark.shuffle.consolidateFiles", "true")
  3. spark.shuffle.consolidateFiles：是否开启shuffle block file的合并，默认为false
  4. spark.reducer.maxSizeInFlight：reduce task的拉取缓存，默认48m
  5. spark.shuffle.file.buffer：map task的写磁盘缓存，默认32k
  6. spark.shuffle.io.maxRetries：拉取失败的最大重试次数，默认3次
  7. spark.shuffle.io.retryWait：拉取失败的重试间隔，默认5s
  8. spark.shuffle.memoryFraction：用于reduce端聚合的内存比例，默认0.2，超过比例就会溢出到磁盘上