Spark程序性能优化

拼搏现实的明天。 2022-09-10 09:27 287阅读 0赞

### 性能优化分析 ###

一个计算任务的执行主要依赖于CPU、内存、带宽  
Spark是一个基于内存的计算引擎，所以对它来说，影响最大的可能就是内存，一般我们的任务遇到了性能瓶颈大概率都是内存的问题，当然了CPU和带宽也可能会影响程序的性能，这个情况也不是没有的，只是比较少。  
Spark性能优化，其实主要就是在于对内存的使用进行调优。通常情况下，如果你的Spark程序计算的数据量比较小，并且你的内存足够使用，那么只要网络不至于卡死，一般是不会有大的性能问题的。但是Spark程序的性能问题往往出现在针对大数据量进行计算(比如上亿条数的数据，或者上T规模的数据)，这个时候如果内存分配不合理就会比较慢，所以Spark性能优化，主要是对内存进行优化。

### 内存都去哪了 ###

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

所以把原始文件中的数据转化为内存中的对象之后，占用的内存会比原始文件中的数据要大

那如何预估程序会消耗多少内存呢?  
通过cache方法，可以看到RDD中的数据cache到内存中之后占用多少内存，这样就能看出了 代码如下:

import org.apache.spark.{ SparkConf, SparkContext}
    
    
    object TestMemoryScala { 
    
      	def main(args: Array[String]): Unit = { 
    	    val conf = new SparkConf()
    	    conf.setAppName("CheckpointOpScala")
    	    .setMaster("local")
    	    val sc = new SparkContext(conf)
    	
    	    val dataRDD = sc.textFile("hdfs://bigdata01:9000/hello_10000000.dat").cache()
    	    val count = dataRDD.count()
    	    println(count)
    	    //while循环是为了保证程序不结束，方便在本地查看4040页面中的storage信息
    	    while(true){ 
    	    	;
    	    }
      	}
    }

执行代码，访问localhost的4040端口界面  
这个界面其实就是spark的任务界面，在本地运行任务的话可以直接访问4040界面查看

点击stages可以看到任务的原始输入数据是多大  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16]  
点击storage可以看到将数据加载到内存，生成RDD之后的大小  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
这样我们就能知道这一份数据在RDD中会占用多少内存了，这样在使用的时候，如果想要把数据全部都加载进内存，就需要给这个任务分配这么多内存了，当然了你分配少一些也可以，只不过这样计算效率会变低，因为RDD中的部分数据内存放不下就会放到磁盘了。

### 性能优化方案 ###

一般对Spark进行性能优化可以有以下几种方案：

*  高性能序列化类库
 *  持久化或者checkpoint
 *  JVM垃圾回收调优
 *  提高并行度
 *  数据本地化
 *  算子优化

#### 高性能序列化类库 ####

在任何分布式系统中，序列化都是扮演着一个重要的角色的。  
如果使用的序列化技术，在执行序列化操作的时候很慢，或者是序列化后的数据还是很大，那么会让分布式应用程序的性能下降很多。所以，进行Spark性能优化的第一步，就是进行序列化的性能优化。  
Spark默认会在一些地方对数据进行序列化，如果我们的算子函数使用到了外部的数据(比如Java中的自定义类型)，那么也需要让其可序列化，否则程序在执行的时候是会报错的，提示没有实现序列化，这个一定要注意。

原因是这样的:  
因为Spark的初始化工作是在Driver进程中进行的，但是实际执行是在Worker节点的Executor进程中进行的;  
当Executor端需要用到Driver端封装的对象时，就需要把Driver端的对象通过序列化传输到 Executor端，这个对象就需要实现序列化。  
否则会报错，提示对象没有实现序列化

> 注意了，其实遇到这种没有实现序列化的对象，解决方法有两种

1.  如果此对象可以支持序列化，则将其实现Serializable接口，让它支持序列化
2.  如果此对象不支持序列化，针对一些数据库连接之类的对象，这种对象是不支持序列化的，所以可以把这个代码放到算子内部，这样就不会通过driver端传过去了，它会直接在executor中执行。

Spark对于序列化的便捷性和性能进行了一个取舍和权衡。默认情况下，Spark倾向于序列化的便捷性，使用了Java自身提供的序列化机制——基于 `ObjectInputStream` 和 `ObjectOutputStream` 的序列化机制， 因为这种方式是Java原生提供的，使用起来比较方便

但是Java序列化机制的性能并不高。序列化的速度相对较慢，而且序列化以后的数据，相对来说还是比较大，比较占空间。所以，如果你的Spark应用程序对内存很敏感，那默认的Java序列化机制并不是最好的选择。

Spark实际上提供了两种序列化机制: `Java序列化机制`和`Kryo序列化机制`  
Spark只是默认使用了java这种序列化机制

*  Java序列化机制:默认情况下，Spark使用Java自身的ObjectInputStream和ObjectOutputStream机制进行对象的序列化。只要你的类实现了Serializable接口，那么都是可以序列化的。Java序列化机制的速度比较慢，而且序列化后的数据占用的内存空间比较大，这是它的缺点
 *  Kryo序列化机制: Spark也支持使用Kryo序列化。Kryo序列化机制比Java序列化机制`更快`，而且序列化后的数据占用的空间`更小`，通常比Java序列化的数据占用的空间要小10倍左右。

> Kryo序列化机制之所以不是默认序列化机制的原因:

第一点:因为有些类型虽然实现了Seriralizable接口，但是它也不一定能够被Kryo进行序列化; 第二点:如果你要得到最佳的性能，Kryo还要求你在Spark应用程序中，对所有你需要序列化的类型都进行手工注册，这样就比较麻烦了

如果要使用Kryo序列化机制  
首先要用 `SparkConf` 设置`spark.serializer`的值为`org.apache.spark.serializer.KryoSerializer` ，就是将Spark的序列化器设置为`KryoSerializer` 。这样Spark在进行序列化时，就会使用Kryo进行序列化了。

使用Kryo时针对需要序列化的类，需要预先进行注册，这样才能获得最佳性能——如果不注册的话， Kryo也能正常工作，只是Kryo必须时刻保存类型的全类名，反而占用不少内存。

Spark默认对Scala中常用的类型在Kryo中做了注册，但是，如果在自己的算子中，使用了外部的自定义类型的对象，那么还是需要对其进行注册。

注册自定义的数据类型格式:

conf.registerKryoClasses(...)

> 注意:如果要序列化的自定义的类型，字段特别多，此时就需要对Kryo本身进行优化，因为Kryo内部的缓存可能不够存放那么大的class对象  
> 此时需要调用`SparkConf.set()`方法，设置`spark.kryoserializer.buffer.mb`参数的值，将其调大，默认值为2 ，单位是 MB ，也就是说最大能缓存 2M 的对象，然后进行序列化。可以在必要时将其调大。

**什么场景下适合使用Kryo序列化?**

一般是针对一些自定义的对象，例如我们自己定义了一个对象，这个对象里面包含了几十M，或者上百M的数据，然后在算子函数内部，使用到了这个外部的大对象  
如果默认情况下，让Spark用java序列化机制来序列化这种外部的大对象，那么就会导致序列化速度比较慢，并且序列化以后的数据还是比较大。  
所以，在这种情况下，比较适合使用Kryo序列化类库，来对外部的大对象进行序列化，提高序列化速度，减少序列化后的内存空间占用。

用代码实现一个案例：

import org.apache.spark.storage.StorageLevel
    import org.apache.spark.{ SparkConf, SparkContext}
    
    
    object KryoSerScala { 
    
      	def main(args: Array[String]): Unit = { 
    	    val conf = new SparkConf()
    	    conf.setAppName("KryoSerScala")
    	      .setMaster("local")
    	      //指定使用kryo序列化机制，注意：如果使用了registerKryoClasses，其实这一行设置是可以省略的
    	      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    	      .registerKryoClasses(Array(classOf[Person])) // 注册自定义的数据类型
    	    val sc = new SparkContext(conf)
    	
    	    val dataRDD = sc.parallelize(Array("hello you","hello me"))
    	    val wordsRDD = dataRDD.flatMap(_.split(" "))
    	    val personRDD = wordsRDD.map(word=>Person(word,18)).persist(StorageLevel.MEMORY_ONLY_SER)
    	    personRDD.foreach(println(_))
    	
    	    while(true){ 
    	      ;
    	    }
      	}
    }
    case class Person(name: String,age: Int) extends Serializable

执行任务，然后访问localhost的4040界面  
在界面中可以看到cache的数据大小是 31 字节。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
那我们把kryo序列化设置去掉，使用默认的java序列化看一下效果修改代码，注释掉这两行代码即可

//.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    //.registerKryoClasses(Array(classOf[Person]))

运行任务，再访问4040界面  
发现此时占用的内存空间是 138 字节，比使用kryo的方式内存空间多占用了将近5倍。  
所以从这可以看出来，使用 kryo 序列化方式对内存的占用会降低很多。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 3]  
如果我们只是将spark的序列化机制改为了kryo序列化，但是没有对使用到的自定义类型手工进行注册，那么此时内存的占用会介于前面两种情况之间

修改代码，只注释掉registerKryoClasses这一行代码

.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    //.registerKryoClasses(Array(classOf[Person]))//注册自定义的数据类型

运行任务，再访问4040界面  
发现此时的内存占用为 123 字节，介于前面的 31 字节和 138 字节之间。  
所以从这可以看出来，在使用kryo序列化的时候，针对自定义的类型最好是手工注册一下，否则就算开 启了kryo序列化，性能的提升也是有限的。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 4]

#### 持久化或者checkpoint ####

针对程序中多次被transformation或者action操作的RDD进行持久化操作，避免对一个RDD反复进行计算，再进一步优化，使用Kryo序列化的持久化级别，减少内存占用  
为了保证RDD持久化数据在可能丢失的情况下还能实现高可靠，则需要对RDD执行Checkpoint操作  
具体实现可参考[《Spark：checkpoint介绍》][Spark_checkpoint]

#### JVM垃圾回收调优 ####

由于Spark是基于内存的计算引擎，RDD缓存的数据，以及算子执行期间创建的对象都是放在内存中的，所以针对Spark任务如果内存设置不合理会导致大部分时间都消耗在垃圾回收上

对于垃圾回收来说，最重要的就是调节RDD缓存占用的内存空间，和算子执行时创建的对象占用的内存空间的比例。  
默认情况下，Spark使用每个executor `60%` 的内存空间来缓存RDD，那么只有 `40%` 的内存空间来存放算子执行期间创建的对象

在这种情况下，可能由于内存空间的不足，并且算子对应的task任务在运行时创建的对象过大，那么一旦发现`40%` 的内存空间不够用了，就会触发JVM的垃圾回收操作。因此在极端情况下，垃圾回收操作可能会被频繁触发。

在这种情况下，如果发现垃圾回收频繁发生。那么就需要对这个比例进行调优了，`spark.storage.memoryFraction`参数的默认值是`0.6`

使用`SparkConf().set("spark.storage.memoryFraction", "0.5")`可以进行修改，就是将RDD缓存占用内存空间的比例降低为`50%`，从而提供更多的内存空间来保存task运行时创建的对象。

因此，对于RDD持久化而言，完全可以使用Kryo序列化，加上降低其executor内存占比的方式，来减少其内存消耗。给task提供更多的内存，从而避免task在执行时频繁触发垃圾回收。

我们可以对task的垃圾回收进行监测，在spark的任务执行界面，可以查看每个task执行消耗的时间，以及task gc消耗的时间。

![重新向集群中提交checkpoint的代码，查看spark任务的task指标信息 确保Hadoop集群、yarn的historyserver进程以及spark的historyserver进程是正常运行的][checkpoint_spark_task_ _Hadoop_yarn_historyserver_spark_historyserver]

#### 提高并行度 ####

实际上Spark集群的资源并不一定会被充分利用到，所以要尽量设置合理的并行度，来充分地利用集群的资源，这样才能提高Spark程序的性能。

Spark会自动设置以文件作为输入源的RDD的并行度，依据其大小，比如HDFS，就会给每一个block创建一个partition，也依据这个设置并行度。对于reduceByKey等会发生shuffle操作的算子，会使用并行度最大的父RDD的并行度

可以手动使用`text()`、`parallelize()` 等方法的第二个参数来设置并行度;  
也可以使用`spark.default.parallelism`参数，来设置统一的并行度。  
Spark官方的推荐是，给集群中的每个cpu core设置 `2~3` 个task

下面来举个例子  
在 spark-submit 脚本中给任务设置了5 个executor，每个executor，设置了2个cpu core

spark-submit \
    --master yarn \
    --deploy-mode cluster \
    --executor-memory 1G \
    --num-executors 5 \
    --executor-cores 2 \
    .....

此时，如果我在代码中设置了默认并行度为5

conf.set("spark.default.parallelism","5")

这个参数设置完了以后，也就意味着所有RDD的partition都被设置成了5个，针对RDD的每一个partition，spark会启动一个task来进行计算，所以对于所有的算子操作，都只会创建5个task来处理对应的RDD中的数据。

但是注意了，我们前面在spark-submit脚本中设置了5个executor，每个executor 2个cpu core，所以这个时候spark其实会向yarn集群申请10个cpu core，但是我们在代码中设置了默认并行度为5，只会产生5个task，一个task使用一个cpu core，那也就意味着有5个cpu core是空闲的，这样申请的资源就浪费了一半。

其实最好的情况，就是每个cpu core都不闲着，一直在运行，这样可以达到资源的最大使用率，其实让一个cpu core运行一个task都是有点浪费的，官方也建议让每个cpu core运行2~3个task，这样可以充分压榨CPU的性能

每个task执行的顺序和执行结束的时间很大概率是不一样的，如果正好有10个cpu，运行10个taks，那么某个task可能很快就执行完了，那么这个CPU就空闲下来了，这样资源就浪费了。

所以说官方推荐，给每个cpu分配2~3个task是比较合理的，可以充分利用CPU资源，发挥它最大的价值。

#### 数据本地化 ####

数据本地化对于Spark Job性能有着巨大的影响。如果数据以及要计算它的代码是在一起的，那么性能当然会非常高。但是，如果数据和计算它的代码是分开的，那么其中之一必须到另外一方的机器上。通常来说，移动代码到其它节点，会比移动数据到代码所在的节点，速度要快得多，因为代码比较小。Spark也正是基于这个数据本地化的原则来构建task调度算法的。

数据本地化指的是，数据离计算它的代码有多近。基于数据距离代码的距离，有几种数据本地化级别:

<table> 
 <thead> 
  <tr> 
   <th>数据本地化级别</th> 
   <th>解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>PROCESS_LOCAL</td> 
   <td>进程本地化，性能最好:数据和计算它的代码在同一个JVM进程中</td> 
  </tr> 
  <tr> 
   <td>NODE_LOCAL</td> 
   <td>节点本地化:数据和计算它的代码在一个节点上，但是不在一个JVM进程中</td> 
  </tr> 
  <tr> 
   <td>NO_PREF</td> 
   <td>数据从哪里过来，性能都是一样的，比如从数据库中获取数据</td> 
  </tr> 
  <tr> 
   <td>RACK_LOCAL</td> 
   <td>数据和计算它的代码在一个机架上，数据需要通过网络在节点之间进行传输</td> 
  </tr> 
  <tr> 
   <td>ANY</td> 
   <td>数据可能在任意地方，比如其它网络环境内，或者其它机架上，性能最差</td> 
  </tr> 
 </tbody> 
</table>

Spark倾向使用最好的本地化级别调度task，但这是不现实的  
如果目前我们要处理的数据所在的executor上目前没有空闲的CPU，那么Spark就会放低本地化级别。这时有两个选择:

1.  等待，直到executor上的cpu释放出来，那么就分配task过去;
2.  立即在任意一个其它executor上启动一个task。

Spark默认会等待指定时间，期望task要处理的数据所在的节点上的executor空闲出一个cpu，从而将task分配过去，只要超过了时间，那么Spark就会将task分配到其它任意一个空闲的executor上

可以设置参数， `spark.locality` 系列参数，来调节Spark等待task可以进行数据本地化的时间

*  spark.locality.wait(3000毫秒):默认等待3秒
 *  spark.locality.wait.process:等待指定的时间看能否达到数据和计算它的代码在同一个JVM
 *  spark.locality.wait.node:等待指定的时间看能否达到数据和计算它的代码在一个节点上执行
 *  spark.locality.wait.rack:等待指定的时间看能否达到数据和计算它的代码在一个机架上

例如这个图里面的task，此时的数据本地化级别是最优的`PROCESS_LOCAL`  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 5]

#### 算子优化 ####

算子优化这一块分4个模块讲

*  map vs mapPartitions
 *  foreach vs foreachPartition
 *  repartition的使用
 *  reduceByKey和groupByKey的区别

##### map vs mapPartitions #####

*  **map:** 对RDD中的每个元素进行操作，一次处理一条数据
 *  **mapPartitions:** 对RDD中每个partition进行操作，一次处理一个分区的数据

所以:

*  **map:** 执行 1 次 map算子只处理 1 个元素，如果 partition 中的元素较多，假设当前已经处理了 1000 个元素，在内存不足的情况下，Spark 可以通过GC等方法回收内存(比如将已处理掉的 1000 个元素从内存中回收)。因此， map 操作通常不会导致OOM（out of memory）异常;
 *  **mapPartitions:** 执行 1 次map算子需要接收该 partition 中的所有元素，因此一旦元素很多而内存不足，就容易导致OOM的异常，也不是说一定就会产生OOM异常，只是和map算子对比的话，相对来说容易产生OOM异常

不过一般情况下，mapPartitions 的性能更高;  
初始化操作、数据库链接等操作适合使用 mapPartitions 操作

这是因为:  
假设需要将 RDD 中的每个元素写入数据库中，这时候就应该把创建数据库链接的操作放置在 mapPartitions 中，创建数据库链接这个操作本身就是个比较耗时的，如果该操作放在 map 中执行，将会频繁执行，比较耗时且影响数据库的稳定性。

使用案例：

import org.apache.spark.{ SparkConf, SparkContext}
    import scala.collection.mutable.ArrayBuffer
    
    
    object MapPartitionsOpScala { 
      	def main(args: Array[String]): Unit = { 
    	    val conf = new SparkConf()
    	    conf.setAppName("MapPartitionsOpScala")
    	      .setMaster("local")
    	    val sc = new SparkContext(conf)
    	
    	    //设置分区数量为2
    	    val dataRDD = sc.parallelize(Array(1,2,3,4,5),2)
    	
    	    //map算子一次处理一条数据
    	    /*val sum = dataRDD.map(item=>{ println("==============") item * 2 }).reduce( _ + _)*/
    	
    	    //mapPartitions算子一次处理一个分区的数据
    	    val sum = dataRDD.mapPartitions(it=>{ 
    	      //建议针对初始化链接之类的操作，使用mapPartitions，放在mapPartitions内部
    	      //例如：创建数据库链接，使用mapPartitions可以减少链接创建的次数，提高性能
    	      //注意：创建数据库链接的代码建议放在次数，不要放在Driver端或者it.foreach内部
    	      //数据库链接放在Driver端会导致链接无法序列化，无法传递到对应的task中执行，所以算子在执行的时候会报错
    	      //数据库链接放在it.foreach()内部还是会创建多个链接，和使用map算子的效果是一样的
    	      println("==================")
    	      val result = new ArrayBuffer[Int]()
    	      //这个foreach是调用的scala里面的函数
    	      it.foreach(item=>{ 
    	        result.+=(item * 2)
    	      })
    	      //关闭数据库链接
    	      result.toIterator
    	    }).reduce(_ + _)
    	
    	
    	    println("sum:"+sum)
    	
    	    sc.stop()
      	}
    }

##### foreach vs foreachPartition #####

*  **foreach:** 一次处理一条数据
 *  **foreachPartition:** 一次处理一个分区的数据

foreachPartition的特性和mapPartitions的特性是一样的，唯一的区别就是 mapPartitions 是 transformation 操作(不会立即执行)，foreachPartition是 action 操作(会立即执行)

使用案例如下：

import org.apache.spark.{ SparkConf, SparkContext}
    
    
    object ForeachPartitionOpScala { 
      	def main(args: Array[String]): Unit = { 
    	    val conf = new SparkConf()
    	    conf.setAppName("ForeachPartitionOpScala")
    	      .setMaster("local")
    	    val sc = new SparkContext(conf)
    	    //设置分区数量为2
    	    val dataRDD = sc.parallelize(Array(1,2,3,4,5),2)
    	    //foreachPartition：一次处理一个分区的数据，作用和mapPartitions类似
    	    //唯一的区是mapPartitions是transformation算子，foreachPartition是action算子
    	    dataRDD.foreachPartition(it=>{ 
    	      //在此处获取数据库链接
    	      println("===============")
    	      it.foreach(item=>{ 
    	        //在这里使用数据库链接
    	        println(item)
    	      })
    	      //关闭数据库链接
    	    })
    	    sc.stop()
      	}
    }

##### repartition的使用 #####

对RDD进行重分区，repartition主要有两个应用场景:

1.  可以调整RDD的并行度: 针对个别RDD，如果感觉分区数量不合适，想要调整，可以通过repartition进行调整，分区调整了之后，对应的并行度也就可以调整了
2.  可以解决RDD中数据倾斜的问题: 如果RDD中不同分区之间的数据出现了数据倾斜，可以通过repartition实现数据重新分发，可以均匀分发到不同分区中

使用案例如下：

import org.apache.spark.{ SparkConf, SparkContext}
    
    
    object RepartitionOpScala { 
      	def main(args: Array[String]): Unit = { 
    	    val conf = new SparkConf()
    	    conf.setAppName("RepartitionOpScala")
    	      .setMaster("local")
    	    val sc = new SparkContext(conf)
    	
    	    //设置分区数量为2
    	    val dataRDD = sc.parallelize(Array(1,2,3,4,5),2)
    	
    	    //重新设置RDD的分区数量为3，这个操作会产生shuffle
    	    //也可以解决RDD中数据倾斜的问题
    	    dataRDD.repartition(3)
    	      .foreachPartition(it=>{ 
    	        println("=========")
    	        it.foreach(println(_))
    	      })
    	
    	    //通过repartition可以控制输出数据产生的文件个数
    	    dataRDD.saveAsTextFile("hdfs://bigdata01:9000/rep-001")
    	    dataRDD.repartition(1).saveAsTextFile("hdfs://bigdata01:9000/rep-002")
    	
    	    sc.stop()
      	}
    }

##### reduceByKey和groupByKey的区别 #####

在实现分组聚合功能时这两个算子有什么区别? 看这两行代码

val counts = wordCountRDD.reduceByKey(_ + _)
    val counts = wordCountRDD.groupByKey().map(wc => (wc._1, wc._2.sum))

这两行代码的最终效果是一样的，都是对wordCountRDD中每个单词出现的次数进行聚合统计

那这两种方式在原理层面有什么区别吗?

首先这两个算子在执行的时候都会产生shuffle, 但是:

1.  当采用reduceByKey时，数据在进行shuffle之前会先进行局部聚合
2.  当使用groupByKey时，数据在shuffle之间不会进行局部聚合，会原样进行shuffle

这样的话reduceByKey就减少了shuffle的数据传送，所以效率会高一些。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 6]  
从图中可以看出来reduceByKey在shuffle之前会先对数据进行局部聚合，而groupByKey不会，所以在实现分组聚合的需求中，reduceByKey性能略胜一筹。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/ab4d76b1f39a4cfeba9e9386ea1c224c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/07abeb44d8a047069b49b1140af37449.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/5ab45ab304874f0f98785e6568570160.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/c84dd5241b79411896033f8c278c0008.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220829/4efdb952ad754c0badefb79473f5956b.png
[Spark_checkpoint]: https://jiandansuifeng.blog.csdn.net/article/details/120110481
[checkpoint_spark_task_ _Hadoop_yarn_historyserver_spark_historyserver]: /images/20220829/04c85e9690b54fa98f314227bcf35e9e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220829/7de2561956d34374abbbced23d843891.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 6]: /images/20220829/89148fc71c4e4c82acec3f8c4d5076e3.png