spark调优综述

àì夳堔傛蜴生んèń 2022-05-12 13:10 185阅读 0赞

这里我们主要讲解开发调优，至于参数调优，上节课我配置的参数里也有所说明，这个还是希望大家在真正的运行的时候自己去体会，因为提交的任务量不同，运行的环境不同，参数设置也不通，没有什么特别的规则去遵循，都是一步一步调出来的。

Spark性能优化的第一步，就是要在开发Spark作业的过程中注意和应用一些性能优化的基本原则。开发调优，就是要让大家了解以下一些Spark基本开发原则。

一、开发调优

1.  避免创建重复的RDD

我们在开发一个Spark作业时，首先是基于某个数据源（比如[Hive][]表或HDFS文件）创建一个初始的RDD；接着对这个RDD执行某个算子操作，然后得到下一个RDD；以此类推，循环往复，直到计算出最终我们需要的结果。在这个过程中，多个RDD会通过不同的算子操作（比如map、reduce等）串起来，这个“RDD串”，就是RDD lineage，也就是“RDD的血缘关系链”。

//也就是说，需要对一份数据执行两次算子操作。

//错误的做法：对于同一份数据执行多次算子操作时，创建多个RDD。

//这里执行了两次textFile方法，针对同一个HDFS文件，创建了两个RDD出来，

//然后分别对每个RDD都执行了一个算子操作。

//这种情况下，Spark需要从HDFS上两次加载hello.txt文件的内容，并创建两个单独的RDD；

//第二次加载HDFS文件以及创建RDD的性能开销，很明显是白白浪费掉的。

val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")

rdd1.map(...)

val rdd2 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt")

rdd2.reduce(...)

//正确的用法：对于一份数据执行多次算子操作时，只使用一个RDD。

1.  对多次使用的RDD进行持久化

Spark中对于一个RDD执行多次算子的默认原理是这样的：每次你对一个RDD执行一个算子操作时，都会重新从源头处计算一遍，计算出那个RDD来，然后再对这个RDD执行你的算子操作。这种方式的性能是很差的。

因此对于这种情况，我们的建议是：对多次使用的RDD进行持久化。此时Spark就会根据你的持久化策略，将RDD中的数据保存到内存或者磁盘中。以后每次对这个RDD进行算子操作时，都会直接从内存或磁盘中提取持久化的RDD数据，然后执行算子，而不会从源头处重新计算一遍这个RDD，再执行算子操作。

// 如果要对一个RDD进行持久化，只要对这个RDD调用cache()和persist()即可。

// 正确的做法。

// cache()方法表示：使用非序列化的方式将RDD中的数据全部尝试持久化到内存中。

// 此时再对rdd1执行两次算子操作时，只有在第一次执行map算子时，才会将这个rdd1从源头处计算一次。

// 第二次执行reduce算子时，就会直接从内存中提取数据进行计算，不会重复计算一个rdd。

val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt").cache()

rdd1.map(...)

rdd1.reduce(...)

// persist()方法表示：手动选择持久化级别，并使用指定的方式进行持久化。

// 比如说，StorageLevel.MEMORY\_AND\_DISK\_SER表示，内存充足时优先持久化到内存中，内存不充足时持久化到磁盘文件中。

// 而且其中的\_SER后缀表示，使用序列化的方式来保存RDD数据，此时RDD中的每个partition都会序列化成一个大的字节数组，然后再持久化到内存或磁盘中。

// 序列化的方式可以减少持久化的数据对内存/磁盘的占用量，进而避免内存被持久化数据占用过多，从而发生频繁GC。

val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt").persist(StorageLevel.MEMORY\_AND\_DISK\_SER)

rdd1.map(...)

rdd1.reduce(...)

**Spark的持久化级别**

<table> 
 <tbody> 
 <tr> 
 <td style="background-color:#ffffff;width:162.5pt;"> 持久化级别 </td> 
 <td style="background-color:#ffffff;width:369pt;"> 含义解释 </td> 
 </tr> 
 <tr> 
 <td style="background-color:#ffffff;width:162.5pt;"> MEMORY_ONLY </td> 
 <td style="background-color:#ffffff;width:369pt;"> 使用未序列化的Java对象格式，将数据保存在内存中。如果内存不够存放所有的数据，则数据可能就不会进行持久化。那么下次对这个RDD执行算子操作时，那些没有被持久化的数据，需要从源头处重新计算一遍。这是默认的持久化策略，使用cache()方法时，实际就是使用的这种持久化策略。 </td> 
 </tr> 
 <tr> 
 <td style="background-color:#ffffff;width:162.5pt;"> MEMORY_AND_DISK </td> 
 <td style="background-color:#ffffff;width:369pt;"> 使用未序列化的Java对象格式，优先尝试将数据保存在内存中。如果内存不够存放所有的数据，会将数据写入磁盘文件中，下次对这个RDD执行算子时，持久化在磁盘文件中的数据会被读取出来使用。 </td> 
 </tr> 
 <tr> 
 <td style="background-color:#ffffff;width:162.5pt;"> MEMORY_ONLY_SER </td> 
 <td style="background-color:#ffffff;width:369pt;"> 基本含义同MEMORY_ONLY。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC。 </td> 
 </tr> 
 <tr> 
 <td style="background-color:#ffffff;width:162.5pt;"> MEMORY_AND_DISK_SER </td> 
 <td style="background-color:#ffffff;width:369pt;"> 基本含义同MEMORY_AND_DISK。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC。 </td> 
 </tr> 
 <tr> 
 <td style="background-color:#ffffff;width:162.5pt;"> DISK_ONLY </td> 
 <td style="background-color:#ffffff;width:369pt;"> 使用未序列化的Java对象格式，将数据全部写入磁盘文件中。 </td> 
 </tr> 
 <tr> 
 <td style="background-color:#ffffff;width:162.5pt;"> MEMORY_ONLY_2, MEMORY_AND_DISK_2, 等等. </td> 
 <td style="background-color:#ffffff;width:369pt;"> 对于上述任意一种持久化策略，如果加上后缀_2，代表的是将每个持久化的数据，都复制一份副本，并将副本保存到其他节点上。这种基于副本的持久化机制主要用于进行容错。假如某个节点挂掉，节点的内存或磁盘中的持久化数据丢失了，那么后续对RDD计算时还可以使用该数据在其他节点上的副本。如果没有副本的话，就只能将这些数据从源头处重新计算一遍了。 </td> 
 </tr> 
 </tbody> 
</table>

1.  尽量避免使用shuffle类算子

如果有可能的话，要尽量避免使用shuffle类算子。因为Spark作业运行过程中，最消耗性能的地方就是shuffle过程。shuffle过程，简单来说，就是将分布在集群中多个节点上的同一个key，拉取到同一个节点上，进行聚合或join等操作。比如reduceByKey、join等算子，都会触发shuffle操作。

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

因此在我们的开发过程中，能避免则尽可能避免使用reduceByKey、join、distinct、repartition等会进行shuffle的算子，尽量使用map类的非shuffle算子。这样的话，没有shuffle操作或者仅有较少shuffle操作的Spark作业，可以大大减少性能开销。

1.  使用Kryo优化序列化性能

Spark默认的是使用Java内置的系列化类，虽然可以处理所有自继承java.io.Serializable的类系列化的类，但是其性能不佳，如果这个成为性能瓶颈，可以使用Kryo系列化类。

Spark默认采用Java的序列化器，这里建议采用Kryo序列化提高性能。实测性能最高甚至提高一倍。

Spark之所以不默认使用Kryo序列化，可能的原因是需要对类进行注册。

Java程序中注册很简单：

SparkConf conf = new SparkConf().setAppName(appName);

conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");

conf.registerKryoClasses(new Class\[\]\{YOUR\_SPARK\_CLASS.class\});

[Hive]: http://lib.csdn.net/base/hive