Spark 优化——RDD持久化(cache、persist、checkpoint)的区别及策略选择

偏执的太偏执、 2022-09-10 11:23 283阅读 0赞

**目录**

一、RDD持久化

1.什么时候该使用持久化（缓存）

2. RDD cache & persist 缓存

3. RDD CheckPoint 检查点

4. cache & persist & checkpoint 的特点和区别

特点

区别

二、cache & persist 的持久化级别及策略选择

Spark的几种持久化级别：

1.MEMORY\_ONLY

2.MEMORY\_AND\_DISK

3.MEMORY\_ONLY\_SER

4.MEMORY\_AND\_DISK\_SER

5.DISK\_ONLY

6.MEMORY\_ONLY\_2, MEMORY\_AND\_DISK\_2, 等等

策略选择

--------------------

# 一、RDD持久化 #

## 1.什么时候该使用持久化（缓存） ##

在Spark应用开发中，有时我们希望能多次使用同一个RDD。如果简单地对RDD进行进行调用，每执行一次action操作，Spark每次都会重算RDD以及它的所有依赖，这在迭代算法中消耗格外的大。

如下面这个例子

def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("persist")
        val sc = new SparkContext(sparkConf)
    
        val list = List("hello spark", "hello scala")
        val rdd = sc.makeRDD(list)
        val flatRDD = rdd.flatMap(_.split(" "))
        val mapRDD = flatRDD.map(
          word => {
            // 为了证明 RDD 对象被重用时，会重新读取数据，打印标识（四个core打印四行）
            println("@@@@@@@@@@@")
            (word, 1)
          })
        // 第一次使用 mapRDD 
        val reduceRDD = mapRDD.reduceByKey(_ + _)
        reduceRDD.collect().foreach(println)
        println("********************************")
        // 第二次使用 mapRDD 
        val groupRDD = mapRDD.groupByKey()
        groupRDD.collect().foreach(println)
        sc.stop()
      }
    
    ==============    运行结果： ================
    @@@@@@@@@@@
    @@@@@@@@@@@
    @@@@@@@@@@@
    @@@@@@@@@@@
    (spark,1)
    (scala,1)
    (hello,2)
    ********************************
    @@@@@@@@@@@
    @@@@@@@@@@@
    @@@@@@@@@@@
    @@@@@@@@@@@
    (spark,CompactBuffer(1))
    (scala,CompactBuffer(1))
    (hello,CompactBuffer(1, 1))

可以看到：

mapRDD.reduceByKey(\_ + \_) 和 mapRDD.groupByKey()两个算子先后用到了mapRDD，按照java的思维，println("@@@@@@@@@@@")应该只会执行一次，但从结果上看，显然执行了两次，mapRDD看似被重用了，但是底层并不是重用。

我们知道，RDD是不存数据的，它封装了计算逻辑和数据结构，只存了RDD的依赖关系，所以在执行完mapRDD.reduceByKey(\_ + \_) 再执行mapRDD.groupByKey()的时候，程序会根据mapRDD的依赖关系将前面的计算流程再走一遍，所以，如果一个RDD需要重复使用，那么需要从头再次执行来获取数据，RDD 对象可以重用，但是数据是无法重用的。

这样如果数据量比较大，并且前面的流程比较复杂的情况下，效率会非常低下，如何解决这个问题呢？就是要进行持久化操作（将中间结果mapRDD保存到内存/文件中）。

## 2. RDD cache & persist 缓存 ##

RDD通过cache()或者persist()将前面的计算结果缓存，默认情况下会把数据缓存在JVM的堆内存中。cache() 不需要传参，persist()需要设置持久化级别。

持久化级别有（先列在这，后面会详细讲）：

*  MEMORY\_ONLY
 *  MEMORY\_AND\_DISK
 *  MEMORY\_ONLY\_SER
 *  MEMORY\_AND\_DISK\_SER；
 *  DISK\_ONLY
 *  MEMORY\_ONLY\_2
 *  MEMORY\_AND\_DISK\_2
 *  等等

我们将前面的例子进行改造：

def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("persist")
        val sc = new SparkContext(sparkConf)
    
        val list = List("hello spark", "hello scala")
        val rdd = sc.makeRDD(list)
        val flatRDD = rdd.flatMap(_.split(" "))
        val mapRDD = flatRDD.map(
          word => {
            // 为了证明 RDD对象被重用时，会重新读取数据，打印标识
            println("@@@@@@@@@@@")
            (word, 1)
          })
        // 将中间结果 mapRDD 进行缓存
        // 第一种方式 cache
        mapRDD.cache()
        // 第二种方式 persist
        mapRDD.persist(StorageLevel.DISK_ONLY)
    
        // 第一次使用 mapRDD 
        val reduceRDD = mapRDD.reduceByKey(_ + _)
        reduceRDD.collect().foreach(println)
        println("********************************")
        // 第二次使用 mapRDD 
        val groupRDD = mapRDD.groupByKey()
        groupRDD.collect().foreach(println)
        sc.stop()
      }
    
    ==============    运行结果： ================
    
    @@@@@@@@@@@
    @@@@@@@@@@@
    @@@@@@@@@@@
    @@@@@@@@@@@
    (spark,1)
    (scala,1)
    (hello,2)
    ********************************
    (spark,CompactBuffer(1))
    (scala,CompactBuffer(1))
    (hello,CompactBuffer(1, 1))

从结果上看，println("@@@@@@@@@@@") 只被执行了一次（打印四条是因为我的CPU核数是4），这时候说明第二次使用mapRDD的时候，实现了复用。

什么？cache()和persist()有什么区别？

我们看下cache()的实现：

/**
       * Persist this RDD with the default storage level (`MEMORY_ONLY`).
       */
      def cache(): this.type = persist()
    
      /**
       * Persist this RDD with the default storage level (`MEMORY_ONLY`).
       */
      def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

可以看到，cache()底层就是调用了persist()，并且将持久化级别设置成MEMORY\_ONLY，也就是说 cache() 和 persist(StorageLevel.MEMORY\_ONLY) 是一样的，那么什么时候用cache，什么时候用persist呢？后面策略选择介绍。

来看看另外一种持久化方式：checkpoint()

## 3. RDD CheckPoint 检查点 ##

所谓的检查点其实就是通过将 RDD 中间结果写入磁盘由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。对 RDD 进行 checkpoint 操作并不会马上被执行，必须执行 Action 操作才能触发。

还是前面的例子，看看checkpoint如何使用：

def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("persist")
        val sc = new SparkContext(sparkConf)
        // 因为checkpoint 需要落盘，需要指定checkpoint存放的位置
            sc.setCheckpointDir("hdfs://****")
        val list = List("hello spark", "hello scala")
        val rdd = sc.makeRDD(list)
        val flatRDD = rdd.flatMap(_.split(" "))
        val mapRDD = flatRDD.map(
          word => {
            // 为了证明 RDD对象被重用时，会重新读取数据，打印标识
            println("@@@@@@@@@@@")
            (word, 1)
          })
        // 使用 checkpoint 持久化中间数据
        mapRDD.checkpoint()
        val reduceRDD = mapRDD.reduceByKey(_ + _)
        reduceRDD.collect().foreach(println)
        println("********************************")
        val groupRDD = mapRDD.groupByKey()
        groupRDD.collect().foreach(println)
        sc.stop()
    
    
        /**
         * 运行结果：
         *
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * (spark,1)
         * (scala,1)
         * (hello,2)
         * ********************************
         * (spark,CompactBuffer(1))
         * (scala,CompactBuffer(1))
         * (hello,CompactBuffer(1, 1))
         */
      }

注意：因为checkpoint 写文件，所以需要指定checkpoint存放的位置，一般保存路径都是在HDFS上。那同样是写文件操作，为什么persist不用设置文件存放路径呢？因为persist写的磁盘文件是临时文件，应用执行完成后就会被删除；而checkpoint路径保存的文件是永久存在的，不会随着应用的结束而被删除。

再来看运行结果：

println("\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*") 下方没有出现"@@@@@@@@@@@"，说明mapRDD得到了复用，但是println("\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*")怎么打印了8行"@@@@@@@@@@@"，也就是println("@@@@@@@@@@@")被执行了两次，这是为啥？

为保证数据的安全，一般情况下，checkpoint会独立执行作业，就是说checkpoint会将mapRDD前面的流程全部走一遍，然后将得到的mapRDD写入磁盘，这样跟业务job互不影响，独立运行。

这样的话println("@@@@@@@@@@@")不还得执行两次吗？跟不加持久化结果一样，都要计算两次。

所以为解决这个问题，checkpoint通常会与cache()/persist()一起使用，在执行checkpoint前先将mapRDD进行缓存，在对mapRDD执行checkpoint操作，这样checkpoint直接从内存（或者persist写的临时文件）获取数据，写入HDFS，就省去了前面的计算操作。

改下前面的代码：

def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("persist")
        val sc = new SparkContext(sparkConf)
        // 因为checkpoint 需要落盘，需要指定checkpoint存放的位置
            sc.setCheckpointDir("hdfs://****")
        val list = List("hello spark", "hello scala")
        val rdd = sc.makeRDD(list)
        val flatRDD = rdd.flatMap(_.split(" "))
        val mapRDD = flatRDD.map(
          word => {
            // 为了证明 RDD对象被重用时，会重新读取数据，打印标识
            println("@@@@@@@@@@@")
            (word, 1)
          })
        // 先缓存数据
        mapRDD.cache()
        // 使用 checkpoint 持久化中间数据
        mapRDD.checkpoint()
        val reduceRDD = mapRDD.reduceByKey(_ + _)
        reduceRDD.collect().foreach(println)
        println("********************************")
        val groupRDD = mapRDD.groupByKey()
        groupRDD.collect().foreach(println)
        sc.stop()
    
    
        /**
         * 运行结果：
         *
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * @@@@@@@@@@@
         * (spark,1)
         * (scala,1)
         * (hello,2)
         * ********************************
         * (spark,CompactBuffer(1))
         * (scala,CompactBuffer(1))
         * (hello,CompactBuffer(1, 1))
         */
      }

这时 println("@@@@@@@@@@@") 就只执行一次了。

## 4. cache & persist & checkpoint 的特点和区别 ##

### 特点 ###

*  cache：
    
     *  将数据临时存储在内存中进行数据重用
     *  会在血缘关系中添加新的依赖，一旦出现问题，可以重头读取数据
 *  persist：
    
     *  将数据临时存储在磁盘文件中进行数据重用
     *  因为涉及到磁盘IO，性能较低，但是数据安全
     *  如果作业执行完毕，临时保存的数据文件会丢失
     *  会在血缘关系中添加新的依赖，一旦出现问题，可以重头读取数据
 *  checkpoint：
    
     *  将数据长久的保存在磁盘文件中进行数据重用
     *  因为涉及到磁盘IO，性能较低，但是数据安全
     *  为了保证数据安全，所以一般情况下，会独立执行作业，
     *  即调用检查点的rdd以前的流程都会重新执行一遍，所以效率比较低
     *  为了能够提高效率，一般情况下，是需要和cache联合使用的
     *  执行过程中，会切断血缘关系，重新建立新的血缘关系，
     *  相当于将一个作业的数据源，由原文件切到检查点落盘的文件
     *  下次重读数据时直接从checkpoint文件中读取
     *  checkpoint等同于改变数据源

### 区别 ###

1.  cache/persist缓存只是将数据保存起来，不切断血缘依赖，只是加一条缓存的依赖关系。而checkpoint检查点会切断血缘依赖，checkpoint下游的RDD直接从checkpoint落盘的文件中读取数据，checkpoint等同于改变了数据源。（如下图）
2.  cache/persist缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等容错性强、高可用的文件系统中，可靠性高。
3.  建议对checkpoint的RDD使用cache缓存，这样checkpoint的job只需从cache缓存中读取数据即可，否则需要从头计算一次RDD。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16][]

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

#  二、cache & persist 的持久化级别及策略选择 #

## Spark的几种持久化级别： ##

### 1.MEMORY\_ONLY ###

使用未序列化的Java对象格式，将数据保存在内存中。如果内存不够存放所有的数据，Spark会自动利用最近最少用(LRU)的缓存策略把最老的分区从内存中移除。下一次要用到已经被移除的分区数据时，这些分区需要从源头处重新计算一遍。这是默认的持久化策略，使用cache()方法时，实际就是使用的这种持久化策略。

### 2.MEMORY\_AND\_DISK ###

使用未序列化的Java对象格式，优先尝试将数据保存在内存中。如果内存不够存放所有的数据，Spark会自动利用最近最少用(LRU)的缓存策略把最老的分区从内存中移除，并将数据溢写到磁盘文件中，下次对这个RDD执行算子时，持久化在磁盘文件中的数据会被读取出来使用。

### 3.MEMORY\_ONLY\_SER ###

基本含义同MEMORY\_ONLY。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式会使缓存过程变慢，因为序列化对象也会消耗一些代价，不过这更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC，同时也可以显著减少JVM的GC时间。

### 4.MEMORY\_AND\_DISK\_SER ###

基本含义同MEMORY\_AND\_DISK。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式会使缓存过程变慢，因为序列化对象也会消耗一些代价，不过这更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC，同时也可以显著减少JVM的GC时间。

### 5.DISK\_ONLY ###

使用未序列化的Java对象格式，将数据全部写入磁盘文件中。

### 6.MEMORY\_ONLY\_2, MEMORY\_AND\_DISK\_2, 等等 ###

对于上述任意一种持久化策略，如果加上后缀\_2，代表的是将每个持久化的数据，都复制一份副本，并将副本保存到其他节点上。这种基于副本的持久化机制主要用于进行容错。假如某个节点挂掉，节点的内存或磁盘中的持久化数据丢失了，那么后续对RDD计算时还可以使用该数据在其他节点上的副本。如果没有副本的话，就只能将这些数据从源头处重新计算一遍了。

## 策略选择 ##

默认情况下，性能最高的当然是MEMORY\_ONLY，但前提是你的内存必须足够足够大，可以绰绰有余地存放下整个RDD的所有数据。因为不进行序列化与反序列化操作，就避免了这部分的性能开销；对这个RDD的后续算子操作，都是基于纯内存中的数据的操作，不需要从磁盘文件中读取数据，性能也很高；而且不需要复制一份数据副本，并远程传送到其他节点上。但是这里必须要注意的是，在实际的生产环境中，恐怕能够直接用这种策略的场景还是有限的，如果RDD中数据比较多时（比如几十亿），直接用这种持久化级别，会导致JVM的OOM内存溢出异常。

如果使用MEMORY\_ONLY级别时发生了内存溢出，那么建议尝试使用MEMORY\_ONLY\_SER级别。该级别会将RDD数据序列化后再保存在内存中，此时每个partition仅仅是一个字节数组而已，大大减少了对象数量，并降低了内存占用。这种级别比MEMORY\_ONLY多出来的性能开销，主要就是序列化与反序列化的开销。但是后续算子可以基于纯内存进行操作，因此性能总体还是比较高的。此外，可能发生的问题同上，如果RDD中的数据量过多的话，还是可能会导致OOM内存溢出的异常。

如果纯内存的级别都无法使用，那么建议使用MEMORY\_AND\_DISK\_SER策略，而不是MEMORY\_AND\_DISK策略。因为既然到了这一步，就说明RDD的数据量很大，内存无法完全放下。序列化后的数据比较少，可以节省内存和磁盘的空间开销。同时该策略会优先尽量尝试将数据缓存在内存中，内存缓存不下才会写入磁盘。

通常不建议使用DISK\_ONLY和后缀为\_2的级别：因为完全基于磁盘文件进行数据的读写，会导致性能急剧降低，有时还不如重新计算一次所有RDD。后缀为\_2的级别，必须将所有数据都复制一份副本，并发送到其他节点上，数据复制以及网络传输会导致较大的性能开销，除非是要求作业的高可用性，否则不建议使用。

--------------------

参考博客：([https://www.cnblogs.com/hark0623/p/5533803.html][https_www.cnblogs.com_hark0623_p_5533803.html])

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/789c3c78bae44c8ba4b9741d4c2562b5.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/90fd8b1a404a4d0883762e53413cfc87.png
[https_www.cnblogs.com_hark0623_p_5533803.html]: https://www.cnblogs.com/hark0623/p/5533803.html