Spark 并行度

Posted 2023-05-07

参考技术A

大数据学习交流微信群

Spark 并行度 是指一个 stage 下并行执行的 task 数量，由于一个 task 线程只能执行一个 rdd partition 分区 ，因此，Spark并行度与Spark作业执行性能息息相关。

假定给Spark 作业分配了足够的资源，比如有 50 个 executor ，每个 executor 有 3 个 cpu core ，这也意味最多可以有 50*3=150 个线程同时执行。

如果没有设置并行度，或者设置过小，比如并行度（或者说分区数）为 10 ，那么只会有 10个 线程来并行执行任务，剩余 140 个线程处于空闲状态，不仅造成严重的资源浪费，并且大大降低了作业性能。

换句话说，你的资源虽然分配足够了，但是问题是，并行度没有与资源相匹配，导致你分配下去的资源都浪费掉了。

合理的并行度的设置，应该是要设置的足够大，大到可以完全合理的利用你的集群资源。比如上面的例子，如果调整并行度为 150 ，则是最理想状态，每个线程都得以被利用，每个分区的数据量也大大减少，性能会得到明显提升。但是官方推荐的并行度是 executor * cpu core 的 2-3倍 。这是因为，考虑到有些任务执行时间短暂，线程释放之后可以马上复用，运行下一个任务。当然，如果任务运行时间长，没有空闲线程，多余的分区必须要等待了。

指定并行度（分区数）的方法参见 Java Spark 简单示例（七） RDD分区 分区划分器

大数据之Spark:Spark调优之RDD算子调优

目录

• 7. repartition/coalesce调节并行度
• 8. reduceByKey本地预聚合
• 9. 使用持久化+checkpoint
• 10. 使用广播变量

7. repartition/coalesce调节并行度

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

Spark SQL的并行度不允许用户自己指定，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算子。

repartition 算子使用前后对比图如下：

Spark SQL这一步的并行度和task数量肯定是没有办法去改变了，但是，对于Spark SQL查询出来的RDD，立即使用repartition算子，去重新进行分区，这样可以重新分区为多个partition，从repartition之后的RDD操作，由于不再涉及Spark SQL，因此stage的并行度就会等于你手动设置的值，这样就避免了Spark SQL所在的stage只能用少量的task去处理大量数据并执行复杂的算法逻辑。使用repartition算子的前后对比如上图所示。

8. reduceByKey本地预聚合

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

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。

9. 使用持久化+checkpoint

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

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

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

持久化设置如下：

sc.setCheckpointDir(‘HDFS’)
rdd.cache/persist(memory_and_disk)
rdd.checkpoint

10. 使用广播变量

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

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

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

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

对于多个Task可能会共用的数据可以广播到每个Executor上：

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

广播变量名.value//获取广播变量

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");