Spark八:Spark性能优化

发布时间:2024年01月10日

Spark性能调优

Spark调优的方法,包括RDD使用、文件读取,partition
学习资料:https://mp.weixin.qq.com/s/caCk3mM5iXy0FaXCLkDwYQ

一、Spark调优之RDD算子调优

1.1 RDD复用

在对RDD进行计算时,要避免相同的算子和计算逻辑下对RDD进行重复的计算。

1.2 尽早filter

获取到初始RDD后,应该考虑尽早地过滤掉不需要的数据,进而减少对内存的占用,从而提升Spark作业的运行效率。

1.3 读取大量小文件-用wholeTextFiles

当我们将一个文本文件读取为 RDD 时,输入的每一行都会成为RDD的一个元素。
也可以将多个完整的文本文件一次性读取为一个pairRDD,其中键是文件名,值是文件内容。

// 这种读法,如果是传递目录,则将目录下所有的文件读取作为RDD,文件路径支持通配符
val input:RDD[String] = sc.textFile("dir/*.log") 

使用wholeTextFiles,返回值为RDD[(String, String)],其中Key是文件的名称,Value是文件的内容。举例:

val filesRDD: RDD[(String, String)] =
sc.wholeTextFiles("D:\\data\\files", minPartitions = 3)
val linesRDD: RDD[String] = filesRDD.flatMap(_._2.split("\\r\\n"))
val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(" "))
wordsRDD.map((_, 1)).reduceByKey(_ + _).collect().foreach(println)

1.4 mapPartition和foreachPartition

1.4.1 mapPartitions

map(_...)表示每一个元素
mapPartitions(_...)表示每个分区的数据组成迭代器
普通的map算子对RDD中的每一个元素进行操作,而mapPartitions算子对RDD中每一个分区进行操作。

如果是普通的map算子,假设一个partition有1万条数据,那么map算子中的function要执行1万次,也就是对每个元素进行操作。
如果是mapPartition算子,由于一个task处理一个RDD的partition,那么一个task只会执行一次function,function一次接收所有的partition数据,效率比较高。

比如,当要把RDD中所有数据通过JDBC写入数据,如果使用map算子,那么需要对RDD中的每一个元素都创建一个数据库连接,这样对资源的消耗很大,如果使用mapPartitions,那么针对一个分区的数据,只需要建立一个数据库连接。

mapPartition算子缺点:对于普通的map操作,一次处理一条数据,如果在处理了2000条数据后内存不足,那么可以将已经处理完的2000条数据从内存中垃圾回收掉;但是如果使用mapPartitions算子,但数据量非常大时,function一次处理一个分区的数据,如果一旦内存不足,此时无法回收内存,就可能会OOM,即内存溢出

因此:mapPartitions算子适用于数据量不是特别大的时候,此时使用mapPartitions算子对性能的提升效果还是不错的

在项目中,应该首先估算一下RDD的数据量、每个partition的数据量,以及分配给每个Executor的内存资源,如果资源允许,可以考虑使用mapPartitions算子代替map。

1.4.2 foreachPartition

rdd.foreach(_...)表示每一个元素
rdd.foreachPartition(_...)表示每个分区的数据组成的迭代器

在生产环境中,通常使用foreachPartition算子来完成数据库的写入,通过foreachPartition算子的特性,可以优化写数据库的性能。
同mapPartition,如果使用foreach算子完成数据库的操作,由于foreach算子是遍历RDD的每条数据,因此,每条数据都会建立一个数据库连接,这是对资源的极大浪费,因此,对于写数据库操作,我们应当使用foreachPartition算子。

与mapPartitions算子非常相似,foreachPartition是将RDD的每个分区作为遍历对象,一次处理一个分区的数据。


使用foreachPartition 算子后,可以获得以下的性能提升:

  1. 对于我们写的function函数,一次处理一整个分区的数据;
  2. 对于一个分区内的数据,创建唯一的数据库连接;
  3. 只需要向数据库发送一次SQL语句和多组参数;

在生产环境中,全部都会使用foreachPartition算子完成数据库操作。foreachPartition算子存在一个问题,与mapPartitions算子类似,如果一个分区的数据量特别大,可能会造成OOM,即内存溢出。

1.5 filter+coalesce/repartition(减少分区)

在Spark任务中我们经常会使用filter算子完成RDD中数据的过滤,在任务初始阶段,从各个分区中加载到的数据量是相近的,但是一旦进过filter过滤后,每个分区的数据量有可能会存在较大差异,如下图所示:
在这里插入图片描述

上图有两个问题:

  1. 每个partition的数据量变小,如果按照之前与partition相等的task个数去处理当前数据,有点浪费task的计算资源;
  2. 每个partition的数据量不一样,会导致后面的每个task处理每个partition数据的时候,每个task要处理的数据量不同,这很有可能导致数据倾斜问题。

针对上面的问题,可以使用coalesce算子。

repartitioncoalesce都可以用来进行重分区,其中repartition只是coalesce接口中shuffle为true的简易实现,coalesce默认情况下不进行shuffle,但是可以通过参数进行设置。
假设我们希望将原本的分区个数A通过重新分区变为B,那么有以下几种情况:

  1. A>B:
    • A与B相差不大:此时使用coalesce即可,无需shuffle过程。
    • A与B相差大:可以使用coalesce并且不启用shuffle过程,但是会导致合并过程性能低下,所以推荐设置coalesce的第二个参数为true,即启动shuffle过程。
  2. A<B:
    • 此时使用repartition即可,如果使用coalesce需要将shuffle设置为true,否则coalesce无效。

可以在filter操作之后,使用coalesce算子针对每个partition的数据量各不相同的情况,压缩partition的数量,而且让每个partition的数据量尽量均匀紧凑,以便于后面的task进行计算操作,在某种程度上能够在一定程度上提升性能。

注意:local模式是进程内模拟集群运行,已经对并行度和分区数量有了一定的内部优化,因此不用去设置并行度和分区数量。

1.6 并行度设置

Spark作业中的并行度指各个stage的task的数量。
如果并行度设置不合理而导致并行度过低,会导致资源的极大浪费。
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作业并行度的设置
val conf = new SparkConf().set("spark.default.parallelism", "500")

1.7 repartition/coalesce调节并行度

Spark 中有并行度的调节策略,但是,并行度的设置对于Spark SQL是不生效的,用户设置的并行度只对于Spark SQL以外的所有Spark的stage生效


Spark SQL自己会默认根据hive表对应的HDFS文件的split个数自动设置Spark SQL所在的那个stage的并行度,用户自己通 spark.default.parallelism 参数指定的并行度,只会在没Spark SQL的stage中生效。

由于Spark SQL所在stage的并行度无法手动设置,如果数据量较大,并且此stage中后续的transformation操作有着复杂的业务逻辑,而Spark SQL自动设置的task数量很少,这就意味着每个task要处理为数不少的数据量,然后还要执行非常复杂的处理逻辑。
这就可能表现为第一个有Spark SQL的stage速度很慢,而后续的没有Spark SQL的stage运行速度非常快。

为了解决Spark SQL无法设置并行度和task数量的问题,我们可以使用repartition算子,使用前后对比图如下:
在这里插入图片描述
Spark SQL的并行度和task数量没有办法改变,但是对于Spark SQL查询出来的RDD,立刻使用repartition算子,重新进行分区。从repartition之后的RDD操作,由于不再涉及Spark SQL,因此stage的并行度就会等于手动设置的值,这样就避免了Spark SQL所在的stage只能用少量的task去处理大量数据并执行复杂的算法逻辑。使用repartition算子的前后对比如上图所示。

1.8 reduceByKey本地预聚合

reduceByKey相较于普通的shuffle操作一个显著的特点就是会进行map端的本地聚合,map端会先对本地的数据进行combine操作,然后将数据写入给下个stage的每个task创建的文件中,也就是在map端,对每一个key对应的value,执行reduceByKey算子函数。
使用reduceByKey对性能的提升:

  1. 本地聚合后,在map端的数据量变少,减少了磁盘IO,也减少了对磁盘空间的占用;
  2. 本地聚合后,下一个stage拉取的数据量变少,减少了网络传输的数据量;
  3. 本地聚合后,在reduce端进行数据缓存的内存占用减少;
  4. 本地聚合后,在reduce端进行聚合的数据量减少。

基于reduceByKey的本地聚合特征,我们应该考虑使用reduceByKey代替其他的shuffle算子,例如groupByKey。

groupByKey与reduceByKey的运行原理如下图1和图2所示:
groupByKey不会进行map端的聚合,而是将所有map端的数据shuffle到reduce端,然后在reduce端进行数据的聚合操作。由于reduceByKey有map端聚合的特性,使得网络传输的数据量减小,因此效率要明显高于groupByKey。
在这里插入图片描述

1.9 使用持久化+checkpoint

Spark持久化在大部分情况下是没有问题的,但是有时候会数据丢失,如果数据一旦丢失,就需要对丢失的数据重新及逆行计算,计算后再缓存和使用。为了避免数据的丢失,可以选择对RDD进行checkpoint,也就是将数据据持久化一份到容错的文件系统上(HDFS)

一个RDD缓存并checkpoint后,一旦发现缓存丢失,会优先查看checkpoint数据存不存在。如果有,就会使用checkpoint数据,而不用重新计算。也即是说,checkpoint可以视为cache的保障机制,如果cache失败,就使用checkpoint的数据。

使用checkpoint的优缺点

优点:提高Spark作业的可靠性,一旦缓存出现问题,不需要重新计算数据
缺点:checkpoint时需要将数据写入HDFS等文件系统,对性能消耗较大。

1.10 使用广播变量

默认情况下,task中的算子如果使用了外部变量,每个task会获得一份变量的副本,这就造成了内存的极大消耗。
另一方面,如果后续对RDD进行持久化,可能就无法将RDD写入内存,只能写入磁盘,磁盘IO将会严重消耗性能;
另一方面,task在创建对象的时候,也许会发现堆内存无法存放新创建的对象,这就会导致频繁的GC,GC会导致工作线程停止,进而导致Spark暂停工作一段时间,严重影响Spark性能。

假设当前任务配置了20个Executor,指定500个task,有一个20M的变量被所有task共用,此时会在500个task中产生500个副本,耗费集群10G的内存,如果使用了广播变量, 那么每个Executor保存一个副本,一共消耗400M内存,内存消耗减少了25倍。


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

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

val 广播变量名= sc.broadcast(会被各个Task用到的变量,即需要广播的变量)
广播变量名.value//获取广播变量

1.11 使用Kryo序列化

默认情况下,Spark使用Java的序列化机制。
Java的序列化机制使用方便,不需要额外的配置,在算子中使用的变量实现Serializable接口即可,但是,Java序列化机制的效率不高,序列化速度慢并且序列化后的数据所占用的空间依然较大

Spark官方宣称Kryo序列化机制比Java序列化机制性能提高10倍左右,Spark之所以没有默认使用Kryo作为序列化类库,是因为它不支持所有对象的序列化,同时Kryo需要用户在使用前注册需要序列化的类型,不够方便,但从Spark 2.0.0版本开始,简单类型、简单类型数组、字符串类型的Shuffling RDDs 已经默认使用Kryo序列化方式了

Kryo序列化注册方式代码如下:

public class MyKryoRegistrator implements KryoRegistrator{
  @Override
  public void registerClasses(Kryo kryo){
    kryo.register(StartupReportLogs.class);
  }
}

配置Kryo序列化方式的代码如下:

//创建SparkConf对象
val conf = new SparkConf().setMaster().setAppName()
//使用Kryo序列化库
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");  
//在Kryo序列化库中注册自定义的类集合
conf.set("spark.kryo.registrator", "bigdata.com.MyKryoRegistrator"); 
文章来源:https://blog.csdn.net/eight_Jessen/article/details/135410967
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。