一文看懂Spark中Cache和CheckPoint的区别

梦里梦外; 2024-02-18 04:34 14阅读 0赞

#### 目录 ####

*   *  循循渐进理解
     *  使用Cache或者Persist
     *  CheckPoint
     *  缓存和CheckPoint的区别

### 循循渐进理解 ###

wc.txt数据

hello java
    spark hadoop flume kafka
    hbase kafka flume hadoop

看下面代码会打印多少条-------------------------(RDD2)

import org.apache.spark.rdd.RDD
    import org.apache.spark.{
        SparkConf, SparkContext}
    
    object Cache {
        
      def main(args: Array[String]): Unit = {
        
        val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
        val rdd1: RDD[String] = sc.textFile("src/main/resources/wc.txt")
    
        val rdd2: RDD[String] = rdd1.flatMap(x => {
        
          println("-------------------------")
          x.split(" ")
        })
    
        val rdd3: RDD[(String, Int)] = rdd2.map(x => (x, 1))
    
        val rdd4: RDD[Int] = rdd2.map(x => x.size)
    
        rdd3.collect()
        rdd4.collect()
        
        Thread.sleep(10000000)
      }
    
    
    }

正确答案是6条(**`解释一下wc.txt里面有三行数据，所以flatmap执行一次，会打印三条`**)，因为执行了两个collect()行动算子(action)  
大致流程就是这样，因为rdd2没有缓存，所以要执行两次  
![在这里插入图片描述][1ae8b804e04a489189d7ac9ae103cf27.png]  
![在这里插入图片描述][98f46b0ee9084d9e8b936ae8f485cd71.png]

上述的问题  
1.一个RDD在多个job中重复使用

*  问题：每个job执行的时候，该RDD之前处理布置也会宠物中
 *  使用持久化的好处：可以将该RDD数据持久化后，后续job在执行在执行的时候可以直接获取数据计算，不用重读RDD之前数据处理

2.如果一个job依赖链条长

*  问题：依赖链条太长的时候，如果数据丢失需要重新计算浪费大量的空间
 *  使用持久化的好处：可以直接持久化数据拿来计算，不用重头计算，节省时间

### 使用Cache或者Persist ###

看下面代码会打印多少条-------------------------(RDD2) 使用了Cache

import org.apache.spark.rdd.RDD
    import org.apache.spark.{
        SparkConf, SparkContext}
    
    object Cache {
        
      def main(args: Array[String]): Unit = {
        
        val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
        val rdd1: RDD[String] = sc.textFile("src/main/resources/wc.txt")
    
        val rdd2: RDD[String] = rdd1.flatMap(x => {
        
          println("-------------------------")
          x.split(" ")
        })
        rdd2.cache()
        val rdd3: RDD[(String, Int)] = rdd2.map(x => (x, 1))
    
        val rdd4: RDD[Int] = rdd2.map(x => x.size)
    
        rdd3.collect()
        rdd4.collect()
    
        Thread.sleep(10000000)
      }
    
    }

正确答案是3条  
![在这里插入图片描述][f1f4ccb4d1494a0ea3ee672b32331c03.png]

发现有个绿色点  
![在这里插入图片描述][55919322a8914e15b8bcb76c883293f5.png]  
发现cache存到memory里面  
![在这里插入图片描述][af637c504aee499cbfeeb10c23542e5a.png]  
RDD的持久化分为  
缓存

*  数据保存位置: task所在主机内存/本地磁盘中
 *  数据保存时机: 在缓存所在第一个Job执行过程中进行数据保存
 *  使用: rdd.cache()/rdd.persist()/rdd.persist(StorageLevel.XXXX)
 *  cache与persist的区别
    
     *  cache是只将数据保存在内存中(cache的底层就是persisit())  
        ![在这里插入图片描述][bc094a97ac9a44568c67272673afafca.png]
     *  persist是可以指定将数据保存在内存/磁盘中  
        ![在这里插入图片描述][cb5932620ba0492a81b47299b631dbfb.png]
 *  常用的存储级别:
    
     *  StorageLevel.MEMORY\_ONLY：只将数据保存在内存中,一般用于小数据量场景
     *  StorageLevel.MEMORY\_AND\_DISK：只将数据保存在内存+磁盘中,一般用于大数据量场景

### CheckPoint ###

看下面代码会打印多少条-------------------------(RDD2) 使用了CheckPoint

import org.apache.spark.rdd.RDD
    import org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK
    import org.apache.spark.{
        SparkConf, SparkContext}
    
    object Cache {
        
      def main(args: Array[String]): Unit = {
        
        System.setProperty("HADOOP_USER_NAME", "root")
        val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
        sc.setCheckpointDir("hdfs://hadoop102:8020/sparkss")
    
    
        val rdd1: RDD[String] = sc.textFile("src/main/resources/wc.txt")
    
        val rdd2: RDD[String] = rdd1.flatMap(x => {
        
          println("-------------------------")
          x.split(" ")
        })
        rdd2.checkpoint()
        val rdd3: RDD[(String, Int)] = rdd2.map(x => (x, 1))
    
        val rdd4: RDD[Int] = rdd2.map(x => x.size)
    
        rdd3.collect()
        rdd4.collect()
        rdd4.collect()
    
    
        Thread.sleep(10000000)
      }
    
    }

正确答案是6条，无论你有多少个行动算子，他都是6条，因为在checkpoint rdd所在第一个job执行完成之后,**`会单独触发一个job计算得到rdd数据之后保存。`**

**为什么要用CheckPoint的原因**  
缓存是将数据保存在主机磁盘/内存中,如果服务器宕机数据丢失,需要重新根据依赖关系计算得到数据,需要花费大量时间,所以需要将数据保存在可靠的存储介质HDFS中,避免后续数据丢失重新计算。

*  数据保存位置: HDFS
 *  数据保存时机: 在checkpoint rdd所在第一个job执行完成之后,**`会单独触发一个job计算得到rdd数据之后保存。`**
 *  使用
    
     *  1、设置保存数据的目录: sc.setCheckpointDir(path)
     *  2、保存数据: rdd.checkpoint

checkpoint会单独触发一个job执行得到数据之后保存,所以导致数据重复计算,此时可以搭配缓存使用: rdd.cache() + rdd.checkpoint(这样只会产生3条)

import org.apache.spark.rdd.RDD
    import org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK
    import org.apache.spark.{
        SparkConf, SparkContext}
    
    object Cache {
        
      def main(args: Array[String]): Unit = {
        
        System.setProperty("HADOOP_USER_NAME", "root")
        val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
        sc.setCheckpointDir("hdfs://hadoop102:8020/sparkss")
    
    
        val rdd1: RDD[String] = sc.textFile("src/main/resources/wc.txt")
    
        val rdd2: RDD[String] = rdd1.flatMap(x => {
        
          println("-------------------------")
          x.split(" ")
        })
        rdd2.cache()
        rdd2.checkpoint()
        val rdd3: RDD[(String, Int)] = rdd2.map(x => (x, 1))
    
        val rdd4: RDD[Int] = rdd2.map(x => x.size)
    
        rdd3.collect()
        rdd4.collect()
        rdd4.collect()
    
    
        Thread.sleep(10000000)
      }
    
    }

### 缓存和CheckPoint的区别 ###

1.数据保存位置不一样

*  缓存是将数据保存在task所在主机磁盘/内存中
 *  checkpoint是将数据保存到HDFS

2、数据保存时机不一样

*  缓存是rdd所在第一个Job执行过程中进行数据保存
 *  checkpoint是rdd所在第一个job执行完成之后保存

3、依赖关系是否保留不一样

*  缓存是将数据保存在task所在主机磁盘/内存中,所以服务器宕机数据丢失,需要根据依赖关系重新计算得到数据,所以rdd的依赖不能切除。
 *  checkpoint是将数据保存到HDFS,数据不会丢失,所以rdd的依赖后续就用不到了,会切除。

[1ae8b804e04a489189d7ac9ae103cf27.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/0b8a93a72c684009b820436f2562cd27.png
[98f46b0ee9084d9e8b936ae8f485cd71.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/1d7c1aefeffa43348fe67a1a23c0cac4.png
[f1f4ccb4d1494a0ea3ee672b32331c03.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/6754e74294ab4c2a89170faf78e13f65.png
[55919322a8914e15b8bcb76c883293f5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/0cbd1b19bde94f4aad120796d826f971.png
[af637c504aee499cbfeeb10c23542e5a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/439f8f00dbec4687a9b39ffb5ae896ca.png
[bc094a97ac9a44568c67272673afafca.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/4f40709ce80b4bb0b6c5d493b0ae5801.png
[cb5932620ba0492a81b47299b631dbfb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/18/a755287a9d1944dfa9e161261ae96fd6.png