Spark应用程序开发参数调优深入剖析-Spark商业调优实战

本套系列博客从真实商业环境抽取案例进行总结和分享，并给出Spark商业应用实战指导，请持续关注本套博客。版权声明：本套Spark商业应用实战归作者（秦凯新）所有，禁止转载，欢迎学习。

• Spark商业应用实战-Spark数据倾斜案例测试及调优准则深入剖析
• Spark商业应用实战-Spark资源调度参数调优深入剖析
• Spark商业应用实战-Spark应用程序开发参数调优深入剖析
• Spark商业应用实战-Spark Shuffle 过程参数优化深入剖析

1 Spark内部资源关系

2 Spark运行资源优化配置

./bin/spark-submit \  
    --master yarn-cluster \  
    --num-executors 100 \  
    --executor-memory 6G \ 
    --executor-cores 4 \
    --driver-memory 1G \
    --conf spark.default.parallelism=1000 \
    --conf spark.storage.memoryFraction=0.5 \  
    --conf spark.shuffle.memoryFraction=0.3 \
复制代码

3 Spark 算子调优建议

• 程序开发调优 ：避免创建重复的RDD

  val rdd1 = sc.textFile("hdfs://master01:9000/hello.txt")
    rdd1.map(...)
    val rdd2 = sc.textFile("hdfs://master01:9000/hello.txt")
    rdd2.reduce(...)
  复制代码

需要对名为“hello.txt”的HDFS文件进行一次map操作，再进行一次reduce操作。 也就是说，需要对一份数据执行两次算子操作。 错误的做法：对于同一份数据执行多次算子操作时，创建多个RDD。 这里执行了两次textFile方法，针对同一个HDFS文件，创建了两个RDD出来 ，然后分别对每个RDD都执行了一个算子操作。 这种情况下，Spark需要从HDFS上两次加载hello.txt文件的内容，并创建两个单独的RDD； 第二次加载HDFS文件以及创建RDD的性能开销，很明显是白白浪费掉的。

• 程序开发调优 ：尽可能复用同一个RDD

• 错误的做法： 有一个<long , String>格式的RDD，即rdd1。 接着由于业务需要，对rdd1执行了一个map操作，创建了一个rdd2， 而rdd2中的数据仅仅是rdd1中的value值而已，也就是说，rdd2是rdd1的子集。

  JavaPairRDD<long , String> rdd1 = ... JavaRDD rdd2 = rdd1.map(...)

分别对rdd1和rdd2执行了不同的算子操作。

rdd1.reduceByKey(...)
    rdd2.map(...)
复制代码

• 正确的做法：

rdd2的数据完全就是rdd1的子集而已，却创建了两个rdd，并对两个rdd都执行了一次算子操作。 此时会因为对rdd1执行map算子来创建rdd2，而多执行一次算子操作，进而增加性能开销。 其实在这种情况下完全可以复用同一个RDD。 我们可以使用rdd1，既做reduceByKey操作，也做map操作。

JavaPairRDD<long , String> rdd1 = ...
rdd1.reduceByKey(...)
rdd1.map(tuple._2...)
复制代码

• 程序开发调优 ：对多次使用的RDD进行持久化

  // 正确的做法。
    // cache()方法表示：使用非序列化的方式将RDD中的数据全部尝试持久化到内存中。
    // 此时再对rdd1执行两次算子操作时，只有在第一次执行map算子时，才会将这个rdd1从源头处计算一次。
    // 第二次执行reduce算子时，就会直接从内存中提取数据进行计算，不会重复计算一个rdd。
    
    val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt").cache()
    rdd1.map(...)
    rdd1.reduce(...)
  
  
    正确的做法：
    // 序列化的方式可以减少持久化的数据对内存/磁盘的占用量，进而避免内存被持久化数据占用过多，
    //从而发生频繁GC。
    
    val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")
    .persist(StorageLevel.MEMORY_AND_DISK_SER)
    rdd1.map(...)
    rdd1.reduce(...)
  复制代码

通常不建议使用DISK_ONLY和后缀为_2的级别：因为完全基于磁盘文件进行数据的读写，会导致性能急剧降低，已经网络较大开销

• 程序开发调优 ：尽量避免使用shuffle类算子

如果有可能的话，要尽量避免使用shuffle类算子，最消耗性能的地方就是shuffle过程。

shuffle过程中，各个节点上的相同key都会先写入本地磁盘文件中，然后其他节点需要通过网络传输拉取各个节点上的磁盘文件中的相同key。而且相同key都拉取到同一个节点进行聚合操作时，还有可能会因为一个节点上处理的key过多，导致内存不够存放，进而溢写到磁盘文件中。因此在shuffle过程中，可能会发生大量的磁盘文件读写的IO操作，以及数据的网络传输操作。磁盘IO和网络数据传输也是shuffle性能较差的主要原因。

尽可能避免使用reduceByKey、join、distinct、repartition等会进行shuffle的算子，尽量使用map类的非shuffle算子。
    // 传统的join操作会导致shuffle操作。
    // 因为两个RDD中，相同的key都需要通过网络拉取到一个节点上，由一个task进行join操作。
       val rdd3 = rdd1.join(rdd2)

    // Broadcast+map的join操作，不会导致shuffle操作。
    // 使用Broadcast将一个数据量较小的RDD作为广播变量。

    // 注意，以上操作，建议仅仅在rdd2的数据量比较少（比如几百M，或者一两G）的情况下使用。
    // 因为每个Executor的内存中，都会驻留一份rdd2的全量数据。
    val rdd2Data = rdd2.collect()
    val rdd2DataBroadcast = sc.broadcast(rdd2Data)
    val rdd3 = rdd1.map(rdd2DataBroadcast...)
复制代码

• 程序开发调优 ：使用map-side预聚合的shuffle操作

如果因为业务需要，一定要使用shuffle操作，无法用map类的算子来替代，那么尽量使用可以map-side预聚合的算子 类似于MapReduce中的本地combiner。map-side预聚合之后，每个节点本地就只会有一条相同的key，因为多条相同的key都被聚合起来了。其他节点在拉取所有节点上的相同key时，就会大大减少需要拉取的数据数量，从而也就减少了磁盘IO以及网络传输开销。

建议使用reduceByKey或者aggregateByKey算子来替代掉groupByKey算子

• 程序开发调优 ：使用高性能的算子
• 使用reduceByKey/aggregateByKey替代groupByKey : map-side
• 使用mapPartitions替代普通map : 函数执行频率
• 使用foreachPartitions替代foreach : 函数执行频率
• 使用filter之后进行coalesce操作 : filter后对分区进行压缩
• 使用repartitionAndSortWithinPartitions替代repartition与sort类操作

repartitionAndSortWithinPartitions是Spark官网推荐的一个算子，官方建议，如果需要在repartition重分区之后，还要进行排序，建议直接使用repartitionAndSortWithinPartitions算子

• 程序开发调优 ：广播大变量

有时在开发过程中，会遇到需要在算子函数中使用外部变量的场景（尤其是大变量，比如100M以上的大集合），那么此时就应该使用Spark的广播（Broadcast）功能来提升性能。 默认情况下，Spark会将该变量复制多个副本，通过网络传输到task中，此时每个task都有一个变量副本。如果变量本身比较大的话（比如100M，甚至1G），那么大量的变量副本在网络中传输的性能开销，以及在各个节点的Executor中占用过多内存导致的频繁GC，都会极大地影响性能。 广播后的变量，会保证每个Executor的内存中，只驻留一份变量副本，而Executor中的task执行时共享该Executor中的那份变量副本。

• 程序开发调优 ：使用Kryo优化序列化性能
• 1、在算子函数中使用到外部变量时，该变量会被序列化后进行网络传输。
• 2、将自定义的类型作为RDD的泛型类型时（比如JavaRDD，Student是自定义类型），所有自定义类型对象，都会进行序列化。因此这种情况下，也要求自定义的类必须实现Serializable接口。
• 3、使用可序列化的持久化策略时（比如MEMORY_ONLY_SER），Spark会将RDD中的每个partition都序列化成一个大的字节数组。

Spark默认使用的是 Java 的序列化机制，你可以使用Kryo作为序列化类库，效率要比 Java的序列化机制要高：

// 创建SparkConf对象。
        val conf = new SparkConf().setMaster(...).setAppName(...)
        // 设置序列化器为KryoSerializer。
        conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        // 注册要序列化的自定义类型。
        conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
复制代码

• 程序开发调优 ：优化数据结构

Java中，有三种类型比较耗费内存：

• 1、对象，每个Java对象都有对象头、引用等额外的信息，因此比较占用内存空间。

• 2、字符串，每个字符串内部都有一个字符数组以及长度等额外信息。

• 3、集合类型，比如HashMap、LinkedList等，因为集合类型内部通常会使用一些内部类来封装集合元素，比如Map.Entry

  Spark官方建议，在Spark编码实现中，特别是对于算子函数中的代码，尽量不要使用上述三种数据结构，尽量使用字符串替代对象，使用原始类型（比如Int、Long）替代字符串，使用数组替代集合类型，这样尽可能地减少内存占用，从而降低GC频率，提升性能。