SparkCore之RDD持久化与检查点设置

清疚 2022-12-31 06:25 169阅读 0赞

# SparkCore之RDD持久化与检查点设置 #

## 1. RDD Cache缓存 ##

RDD通过Cache或者Persist方法将前面的计算结果缓存，默认情况下会把数据以序列化的形式缓存在JVM的堆内存中。但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中，并供后面重用。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

1）代码实现

object cache01 { 
    
        def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            //3. 创建一个RDD，读取指定位置文件:hello atguigu atguigu
            val lineRdd: RDD[String] = sc.textFile("input1")
    
            //3.1.业务逻辑
            val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
    
            val wordToOneRdd: RDD[(String, Int)] = wordRdd.map { 
                word => { 
                    println("************")
                    (word, 1)
                }
            }
    
            //3.5 cache操作会增加血缘关系，不改变原有的血缘关系
            println(wordToOneRdd.toDebugString)
    
            //3.4 数据缓存。
            wordToOneRdd.cache()
    
            //3.6 可以更改存储级别
            // wordToOneRdd.persist(StorageLevel.MEMORY_AND_DISK_2)
    
            //3.2 触发执行逻辑
            wordToOneRdd.collect()
    
            println("-----------------")
            println(wordToOneRdd.toDebugString)
    
            //3.3 再次触发执行逻辑
            wordToOneRdd.collect()
    
            Thread.sleep(1000000)
    
            //4.关闭连接
            sc.stop()
        }
    }

2）源码解析

mapRdd.cache()
    def cache(): this.type = persist()
    def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)
    
    object StorageLevel { 
      val NONE = new StorageLevel(false, false, false, false)
      val DISK_ONLY = new StorageLevel(true, false, false, false)
      val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
      val MEMORY_ONLY = new StorageLevel(false, true, false, true)
      val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
      val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
      val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
      val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
      val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
      val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
      val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
      val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

注意：默认的存储级别都是仅在内存存储一份。在存储级别的末尾加上“\_2”表示持久化的数据存为两份。SER：表示序列化。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

> 缓存有可能丢失，或者存储于内存的数据由于内存不足而被删除，RDD的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列转换，丢失的数据会被重算，由于RDD的各个Partition是相对独立的，因此只需要计算丢失的部分即可，并不需要重算全部Partition。

3）自带缓存算子

> Spark会自动对一些Shuffle操作的中间数据做持久化操作（比如：reduceByKey）。这样做的目的是为了当一个节点Shuffle失败了避免重新计算整个输入。但是，在实际使用的时候，如果想重用数据，仍然建议调用persist或cache。

object cache02 { 
    
        def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            //3. 创建一个RDD，读取指定位置文件:hello atguigu atguigu
            val lineRdd: RDD[String] = sc.textFile("input1")
    
            //3.1.业务逻辑
            val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
    
            val wordToOneRdd: RDD[(String, Int)] = wordRdd.map { 
                word => { 
                    println("************")
                    (word, 1)
                }
            }
    
            // 采用reduceByKey，自带缓存
            val wordByKeyRDD: RDD[(String, Int)] = wordToOneRdd.reduceByKey(_+_)
    
            //3.5 cache操作会增加血缘关系，不改变原有的血缘关系
            println(wordByKeyRDD.toDebugString)
    
            //3.4 数据缓存。
            //wordByKeyRDD.cache()
    
            //3.2 触发执行逻辑
            wordByKeyRDD.collect()
    
            println("-----------------")
            println(wordByKeyRDD.toDebugString)
    
            //3.3 再次触发执行逻辑
            wordByKeyRDD.collect()
    
            Thread.sleep(1000000)
    
            //4.关闭连接
            sc.stop()
        }
    }

访问http://localhost:4040/jobs/页面，查看第一个和第二个job的DAG图。说明：增加缓存后血缘依赖关系仍然有，但是，第二个job取的数据是从缓存中取的。  
![在这里插入图片描述][20201228210841254.png]

## 2. RDD CheckPoint检查点 ##

**1）检查点**：是通过将RDD中间结果写入磁盘。

**2）为什么要做检查点？**

由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。

**3）检查点存储路径**：Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统

**4）检查点数据存储格式为**：二进制的文件

**5）检查点切断血缘**：在Checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。

6）检查点触发时间：对RDD进行Checkpoint操作并不会马上被执行，必须执行Action操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]

7）设置检查点步骤

（1）设置检查点数据存储路径：sc.setCheckpointDir("./checkpoint1")  
（2）调用检查点方法：wordToOneRdd.checkpoint()

8）代码实现

object checkpoint01 { 
    
        def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            // 需要设置路径，否则抛异常：Checkpoint directory has not been set in the SparkContext
            sc.setCheckpointDir("./checkpoint1")
    
            //3. 创建一个RDD，读取指定位置文件:hello atguigu atguigu
            val lineRdd: RDD[String] = sc.textFile("input1")
    
            //3.1.业务逻辑
            val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
    
            val wordToOneRdd: RDD[(String, Long)] = wordRdd.map { 
                word => { 
                    (word, System.currentTimeMillis())
                }
            }
    
            //3.5 增加缓存，避免再重新跑一个job做checkpoint
    // wordToOneRdd.cache()
    
            //3.4 数据检查点：针对wordToOneRdd做检查点计算
            wordToOneRdd.checkpoint()
    
            //3.2 触发执行逻辑
            wordToOneRdd.collect().foreach(println)
            // 会立即启动一个新的job来专门的做checkpoint运算
    
            //3.3 再次触发执行逻辑
            wordToOneRdd.collect().foreach(println)
            wordToOneRdd.collect().foreach(println)
    
            Thread.sleep(10000000)
    
            //4.关闭连接
            sc.stop()
        }
    }

9）执行结果  
访问http://localhost:4040/jobs/页面，查看4个job的DAG图。其中第2个图是checkpoint的job运行DAG图。第3、4张图说明，检查点切断了血缘依赖关系。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 3]

（1）只增加checkpoint，没有增加Cache缓存打印

> 第1个job执行完，触发了checkpoint，第2个job运行checkpoint，并把数据存储在检查点上。第3、4个job，数据从检查点上直接读取。  
> (hadoop,1577960215526)  
> 。。。。。。  
> (hello,1577960215526)  
> (hadoop,1577960215609)  
> 。。。。。。  
> (hello,1577960215609)  
> (hadoop,1577960215609)  
> 。。。。。。  
> (hello,1577960215609)

（2）增加checkpoint，也增加Cache缓存打印

> 第1个job执行完，数据就保存到Cache里面了，第2个job运行checkpoint，直接读取Cache里面的数据，并把数据存储在检查点上。第3、4个job，数据从检查点上直接读取。  
> (hadoop,1577960642223)  
> 。。。。。。  
> (hello,1577960642225)  
> (hadoop,1577960642223)  
> 。。。。。。  
> (hello,1577960642225)  
> (hadoop,1577960642223)  
> 。。。。。。  
> (hello,1577960642225)  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 4]

## 3. 缓存和检查点区别 ##

1）Cache缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。

2）Cache缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等容错、高可用的文件系统，可靠性高。

3）建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据即可，否则需要再从头计算一次RDD。

4）如果使用完了缓存，可以通过unpersist()方法释放缓存

## 4. 检查点存储到HDFS集群 ##

如果检查点数据存储到HDFS集群，要注意配置访问集群的用户名。否则会报访问权限异常。

object checkpoint02 { 
    
        def main(args: Array[String]): Unit = { 
    
            // 设置访问HDFS集群的用户名
            System.setProperty("HADOOP_USER_NAME","hadoop")
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            // 需要设置路径.需要提前在HDFS集群上创建/checkpoint路径
            sc.setCheckpointDir("hdfs://hadoop102:8020/checkpoint")
    
            //3. 创建一个RDD，读取指定位置文件:hello atguigu atguigu
            val lineRdd: RDD[String] = sc.textFile("input1")
    
            //3.1.业务逻辑
            val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
    
            val wordToOneRdd: RDD[(String, Long)] = wordRdd.map { 
                word => { 
                    (word, System.currentTimeMillis())
                }
            }
    
            //3.4 增加缓存，避免再重新跑一个job做checkpoint
            wordToOneRdd.cache()
    
            //3.3 数据检查点：针对wordToOneRdd做检查点计算
            wordToOneRdd.checkpoint()
    
            //3.2 触发执行逻辑
            wordToOneRdd.collect().foreach(println)
    
            //4.关闭连接
            sc.stop()
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221120/9411b1111eac4d1aac58ff9364d5fafd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221120/b2255f635bda40088bc75b1f8ffd905d.png
[20201228210841254.png]: /images/20221120/c61b71f8aca64b38aaa599ecf832a5a9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]: /images/20221120/6e0f4be85f2f4e8ba51af07059be74d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 3]: /images/20221120/99158d295ef74d10a19027ce349e9781.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 4]: /images/20221120/76bfc34beb084bed8628bc01a3cd35c8.png