RDD持久化

作者: 万事万物 | 来源:发表于2021-07-14 09:31 被阅读0次

为什么需要持久化

所谓的持久化，就是将数据进行保存，避免数据丢失。RDD持久化并非将数据落盘保存，而是用作缓存。
了解RDD持久化前需要先了解什么是RDD？

RDD就像是一个水管，数据就像是水，水只会经过水管，并不是存储水。所以RDD是不会存储数据的。
RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，的其中一个特性就是弹性
- 存储的弹性: spark计算过程中中间结果会保存在内存中如果内存不足会自动存储在磁盘
- 容错的弹性: spark中计算过程中如果出错会自动重试
- 计算的弹性：如果计算过程中数据丢失,会根据RDD的依赖关系重新计算得到数据 $\color{red}{(主要说这个)}$
- 分区的弹性：spark RDD会根据文件大小动态分区

扯到弹性，一是该篇文章会讲到，最主要的还是我需要复习一下
这里主要说到计算的弹性，如果其中一个RDD出现问题，可以根据依赖关系重写计算，获得结果。这样的好处就是保证了数据的完整性。那么有没有缺点呢？答案是有的，就是需要重复计算，得从头开始。

1001
如图（1001）：若rdd5出现问题了，若要重新获得数据需要从rdd1开始运行，但是rdd5实际上只依赖于rdd4产生的数据结果。从头开始效率大大降低呀，我们是否可以把rdd4的解决结果缓存起来，rdd5直接从缓存中获取？若缓存中没有在从头开始呢？这样可以大大减少运行时间。

我们再通过一个案例深入理解缓存的作用。

  @Test
  def check2Test(): Unit ={
    val conf =new SparkConf().setMaster("local[4]").setAppName("test")
    val sc=new SparkContext(conf)

    val rdd1=sc.parallelize(List(1,2,3),4)

    //进行一个map算子操作
    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })

    //进行一个reduce算子操作
    val rdd3=rdd2.map(x=>x+10)
    val rdd4=rdd2.map(x=>x+10)


    //打印 rdd2 的结果
    println(rdd3.collect.toList)
    // 输出 sum值
    println(rdd4.collect.toList)
  }

有这么一个程序（如上），rdd2的结果会作为rdd3和rdd4的数据来源，也就是说rdd3和rdd4会对rdd2的结果做运算。
若此时运行程序;rdd2中的println("*"*10)会被运行多少次？

    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })

答案会被运行6次，因为会经过两次collect，
根据依赖关系，调用类似collect 的Action运行算子，程序会从上往下运行（三条数据*两次collect=6次）。

**********
**********
**********
**********
**********
**********

你可以发现到了问题所在，这样每次都要从头开始，这种依赖关系比较少只有三级，若在实际开发中，十级以上都是属于正常的依赖关系，若在rdd2中存有数据，就无需从头开始了，直接取来用即可。

RDD缓存

RDD不存储数据,所以默认情况下每次执行的时候都会stage开头执行
缓存:

数据保存位置: 保存在task所在主机的内存/本地磁盘上
应用场景: 某个RDD在多个job中重复使用的时候

如何缓存:

cache
persist

缓存的好处:

如果一个RDD有设置cache\persist，此时rdd所属第一个Job执行完成之后,数据会持久化到本地的磁盘/内存中。后续RDD所属的其他job在执行的时候会直接将缓存数据拿过来使用而不用重新计算

RDD Cache缓存

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

如何使用？
语法: RDD.cache()

@Test
  def check2Test(): Unit ={
    val conf =new SparkConf().setMaster("local[4]").setAppName("test")
    val sc=new SparkContext(conf)

    val rdd1=sc.parallelize(List(1,2,3),4)

    //进行一个map算子操作
    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })
    //添加 rdd2的缓存
    val rddx=rdd2.cache()

    //进行一个reduce算子操作
    val rdd3=rddx.map(x=>x+10)
    val rdd4=rddx.map(x=>x+10)


    //打印 rdd2 的结果
    println(rdd3.collect.toList)
    // 输出 sum值
    println(rdd4.collect.toList)

  }

运行结果，是否还会打印6次println("*"*10)

**********
**********
**********

数据直接从缓存中取，所以只会打印三次。

cache流程图

RDD Persist缓存

如何使用？
语法: RDD.persist()

@Test
  def check2Test(): Unit ={
    val conf =new SparkConf().setMaster("local[4]").setAppName("test")
    val sc=new SparkContext(conf)

    val rdd1=sc.parallelize(List(1,2,3),4)

    //进行一个map算子操作
    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })
    //添加 rdd2的缓存
    val rddx=rdd2.persist()

    //进行一个reduce算子操作
    val rdd3=rddx.map(x=>x+10)
    val rdd4=rddx.map(x=>x+10)


    //打印 rdd2 的结果
    println(rdd3.collect.toList)
    // 输出 sum值
    println(rdd4.collect.toList)

  }

结果还是3次

**********
**********
**********

Cache和Persist有什么区别？

cache底层就是persist，调用是一个无参函数。

 def cache(): this.type = persist()

无参的persist()其实调用的就是有参的，指定一个缓存级别。

def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

cache是将数据保存在本地内存中
persist可以指定将数据保存在内存/磁盘中

Persist的缓存级别

StorageLevel 中就定义persist的用到缓存级别。

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)

存储级别:

NONE：不存储
DISK_ONLY : 只保存在磁盘中
DISK_ONLY_2 : 只保存在磁盘中,数据保存两份
MEMORY_ONLY ：只保存在内存中
MEMORY_ONLY_2 ：只保存在内存中,数据保存两份
MEMORY_ONLY_SER ：只保存在内存中,以序列化形式存储
MEMORY_ONLY_SER_2 ：只保存在内存中,以序列化形式存储，数据保存两份
MEMORY_AND_DISK ：数据保存在内存/磁盘中,可以动态调整
MEMORY_AND_DISK_2 ：数据保存在内存/磁盘中,可以动态调整，数据保存两份
MEMORY_AND_DISK_SER ：数据保存在内存/磁盘中,可以动态调整,以序列化形式存储
MEMORY_AND_DISK_SER_2 ：数据保存在内存/磁盘中,可以动态调整,以序列化形式存储，数据保存两份
OFF_HEAP ：数据保存在堆外内存中

太多了对不对？其中工作中常用的存储级别:

MEMORY_ONLY<只适用于小数据量场景>
MEMORY_AND_DISK<适用于大数据量场景>

RDD CheckPoint检查点

虽然我们配置了缓存但是只能保存在task所在主机的内存/本地磁盘上，若该服务器出现问题，依然会造成数据丢失，从头开始计算，效率也有所降低。为了安全起见，我们可以设置CheckPoint，将数据保存到比较可靠的地方，如：将数据保存到HDFS中。

1）检查点：是通过将RDD中间结果写入磁盘。
2）为什么要做检查点？
由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。
3）检查点存储路径：Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统
4）检查点数据存储格式为：二进制的文件
5）检查点切断血缘：在Checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。
6）检查点触发时间：对RDD进行Checkpoint操作并不会马上被执行，必须执行Action操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍。

案例演示：

  @Test
  def check4Test(): Unit ={
    val conf =new SparkConf().setMaster("local[4]").setAppName("test")
    val sc=new SparkContext(conf)
    //设置存储路径
    sc.setCheckpointDir("hdfs://hadoop102:9820/output/a")


    val rdd1=sc.parallelize(List(1,2,3),4)

    //进行一个map算子操作
    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })
    //添加 rdd2的缓存
    rdd2.checkpoint()

    //进行一个reduce算子操作
    val rdd3=rdd2.map(x=>x+10)
    val rdd4=rdd2.map(x=>x+10)

    //打印 rdd2 的结果
    println(rdd3.collect.toList)
    // 输出 sum值
    println(rdd4.collect.toList)

  }

/output/a 输出文件

CheckPoint输出文件

若需要这种错误

Permission denied: user=123456, access=WRITE, inode="/output/a":hadoop:supergroup:drwxr-xr-x

这种属于权限问题，解决方式设置-DHADOOP_USER_NAME=hadoop
hadoop=你链接服务器的账号

配置-DHADOOP_USER_NAME

奇怪的是控制台却打印了6次println("*"*10)

**********
**********
**********
**********
**********
**********

这是怎么回事了？
是我执行了 rdd3和rdd4的原因？

  @Test
  def check4Test(): Unit ={
    val conf =new SparkConf().setMaster("local[4]").setAppName("test")
    val sc=new SparkContext(conf)
    //设置存储路径
    sc.setCheckpointDir("hdfs://hadoop102:9820/output/a")


    val rdd1=sc.parallelize(List(1,2,3),4)

    //进行一个map算子操作
    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })
    //添加 rdd2的缓存
    rdd2.checkpoint()

    //进行一个reduce算子操作
    val rdd3=rdd2.map(x=>x+10)
    val rdd4=rdd2.map(x=>x+10)
    val rdd5=rdd2.map(x=>x+10)


    //打印 rdd2 的结果
    println(rdd3.collect.toList)
    // 输出 rdd4 的结果
    println(rdd4.collect.toList)
    // 输出 rdd5 的结果
    println(rdd5.collect.toList)

  }

在加一个 rdd5 是否会执行9次？
启动运行依旧是6次

**********
**********
**********
**********
**********
**********

啊？这么奇怪？那么把rdd4和rdd5删了呢？

  @Test
  def check4Test(): Unit ={
    val conf =new SparkConf().setMaster("local[4]").setAppName("test")
    val sc=new SparkContext(conf)
    //设置存储路径
    sc.setCheckpointDir("hdfs://hadoop102:9820/output/a")


    val rdd1=sc.parallelize(List(1,2,3),4)

    //进行一个map算子操作
    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })
    //添加 rdd2的缓存
    rdd2.checkpoint()

    //进行一个reduce算子操作
    val rdd3=rdd2.map(x=>x+10)


    //打印 rdd2 的结果
    println(rdd3.collect.toList)

  }

启动运行依旧是6次

**********
**********
**********
**********
**********
**********

这是什么原因呢？这就得从checkpoint() 源码中寻址答案了。

def checkpoint(): Unit = RDDCheckpointData.synchronized {
    // 判断路径是否为空
    if (context.checkpointDir.isEmpty) {
      throw new SparkException("Checkpoint directory has not been set in the SparkContext")
    } else if (checkpointData.isEmpty) {
      // 看看 ReliableRDDCheckpointData
      checkpointData = Some(new ReliableRDDCheckpointData(this))
    }
  }

ReliableRDDCheckpointData 是一个伴生对象，我们看看doCheckpoint方法

  protected override def doCheckpoint(): CheckpointRDD[T] = {
    // 将RDD写入检查点目录
    val newRDD = ReliableCheckpointRDD.writeRDDToCheckpointDirectory(rdd, cpDir)

    // Optionally clean our checkpoint files if the reference is out of scope
    if (rdd.conf.get(CLEANER_REFERENCE_TRACKING_CLEAN_CHECKPOINTS)) {
      rdd.context.cleaner.foreach { cleaner =>
        cleaner.registerRDDCheckpointDataForCleanup(newRDD, rdd.id)
      }
    }

    logInfo(s"Done checkpointing RDD ${rdd.id} to $cpDir, new parent is RDD ${newRDD.id}")
    newRDD
  }

进入到 writeRDDToCheckpointDirectory 中

def writeRDDToCheckpointDirectory[T: ClassTag](
      originalRDD: RDD[T],
      checkpointDir: String,
      blockSize: Int = -1): ReliableCheckpointRDD[T] = {
    val checkpointStartTimeNs = System.nanoTime()

    val sc = originalRDD.sparkContext

    // Create the output path for the checkpoint
    val checkpointDirPath = new Path(checkpointDir)
    val fs = checkpointDirPath.getFileSystem(sc.hadoopConfiguration)
    if (!fs.mkdirs(checkpointDirPath)) {
      throw new SparkException(s"Failed to create checkpoint path $checkpointDirPath")
    }

    // Save to file, and reload it as an RDD
    //保存到文件，并将其作为RDD重新加载
    val broadcastedConf = sc.broadcast(
      new SerializableConfiguration(sc.hadoopConfiguration))
    // 上面都是一些初始化工作，判断，创建目录啥，主要是这一句，会在运行job
    sc.runJob(originalRDD, writePartitionToCheckpointFile[T](checkpointDirPath.toString, broadcastedConf) _)
    // 下面就不是很重要了
    if (originalRDD.partitioner.nonEmpty) {
      writePartitionerToCheckpointDir(sc, originalRDD.partitioner.get, checkpointDirPath)
    }

    val checkpointDurationMs =
      TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - checkpointStartTimeNs)
    logInfo(s"Checkpointing took $checkpointDurationMs ms.")

    val newRDD = new ReliableCheckpointRDD[T](
      sc, checkpointDirPath.toString, originalRDD.partitioner)
    if (newRDD.partitions.length != originalRDD.partitions.length) {
      throw new SparkException(
        "Checkpoint RDD has a different number of partitions from original RDD. Original " +
          s"RDD [ID: ${originalRDD.id}, num of partitions: ${originalRDD.partitions.length}]; " +
          s"Checkpoint RDD [ID: ${newRDD.id}, num of partitions: " +
          s"${newRDD.partitions.length}].")
    }
    newRDD
  }

程序运行到这里，他会再执行一个job任务，并将数据保存到缓存中。

sc.runJob(originalRDD, writePartitionToCheckpointFile[T](checkpointDirPath.toString, broadcastedConf) _)

checkpoint一开始就执行(如下），虽然checkpoint在rdd3前面被调用，但是它需要等到rdd3执行完collect之后再运行一个job。

   //添加 rdd2的缓存
    rdd2.checkpoint()

    //进行一个reduce算子操作
    val rdd3=rdd2.map(x=>x+10)

总结：checkpoint会触发一次job操作,该job操作是在checkpoint所属RDD第一个job执行完成之后才会触发；

这样就不爽了，明明只需要跑一次的任务就可以缓存的，现在需要多跑一次，虽然对后面的RDD效率有所提高，但是就是不爽。
其实check可以和checkpoint 配合使用

  @Test
  def check4Test(): Unit ={
    val conf =new SparkConf().setMaster("local[4]").setAppName("test")
    val sc=new SparkContext(conf)
    //设置存储路径
    sc.setCheckpointDir("hdfs://hadoop102:9820/output/a")


    val rdd1=sc.parallelize(List(1,2,3),4)

    //进行一个map算子操作
    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })
    //添加 rdd2的缓存
    val rddx=rdd2.cache()
    rddx.checkpoint()

    //进行一个reduce算子操作
    val rdd3=rddx.map(x=>x+10)
    val rdd4=rddx.map(x=>x+10)
    val rdd5=rddx.map(x=>x+10)


    //打印 rdd2 的结果
    println(rdd3.collect.toList)
    // 输出 rdd4 的结果
    println(rdd4.collect.toList)
    // 输出 rdd5 的结果
    println(rdd5.collect.toList)

    //释放
    rddx.unpersist(true)
    //关闭链接
    sc.stop()

  }

这样配置配置之后，就只会运行一次了。

//添加 rdd2的缓存
val rddx=rdd2.cache()
rddx.checkpoint()

程序结果

**********
**********
**********

注意：checkpoint还是会运行一个job，但是程序不用从头开始了，而是直接从rddx中取。
总结：为了避免checkpoint触发的job重复执行之前的数据处理逻辑,可以在checkpoint之间将rdd通过cache缓存数据,后续checkpoint触发的job就可以直接使用缓存的数据

最后还得注意的是，缓存是可以备份到内存或磁盘中的。
使用cache时，job结束之后，缓存会被自动释放。
使用checkpoint时，需要手动进行释放，需要设置unpersist为true默认为false。

    //释放
    rddx.unpersist(true)
    //关闭链接
    sc.stop()

cache与checkpoint的区别:

数据持久化的位置不一样:

cache是将数据保存在本地内存/磁盘中
checkpoint是将数据保存在HDFS中

依赖关系是否保留不一样

cache是将数据保存在本地内存/磁盘中,如果服务器宕机,此时数据丢失,丢失数据之后只能根据rdd的依赖关系重新计算得到数据,所以rdd的依赖关系会保留
checkpoint是将数据保存在HDFS,数据不会丢失,此时RDD的依赖关系会切除

工作常用还是cache

1）Cache缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。
2）Cache缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等容错、高可用的文件系统，可靠性高。
3）建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据即可，否则需要再从头计算一次RDD。
4）如果使用完了缓存，可以通过unpersist()方法释放缓存

Shuffle会自动进行缓存

Job程序运行中，若进行了shuffle操作，会自动完成一次缓存操作。

案例演示：

  @Test
  def check4Test(): Unit ={
    val conf =new SparkConf().setMaster("local[4]").setAppName("test")
    val sc=new SparkContext(conf)


    val rdd1=sc.parallelize(List(1,2,3),4)

    //进行一个map算子操作
    val rdd2=rdd1.map(x=>{
      println("*"*10)
      x*10
    })

    // 进行一次分组，主要想让程序完成一次shuffle操作。
    val rddx=rdd2.groupBy(x=>x)

    //进行一个reduce算子操作
    val rdd3=rddx.map(x=>x)
    val rdd4=rddx.map(x=>x)
    val rdd5=rddx.map(x=>x)


    //打印 rdd2 的结果
    println(rdd3.collect.toList)
    // 输出 rdd4 的结果
    println(rdd4.collect.toList)
    // 输出 rdd5 的结果
    println(rdd5.collect.toList)

    //关闭链接
    sc.stop()

  }

将代码做了一些变动；

删除所有的缓存
使用groupBy，让程序完成一次shuffle操作
最终看程序，是否只打印 3次println("*"*10)?

运行结果：只打印了3次。

**********
**********
**********

这个一定要记住，面试可能会被问到。

最后

文章内容，我一边看一边做案例，可能文档会比较乱，没有其他博客那么清晰明了。

RDD持久化

为什么需要持久化

RDD缓存

RDD Cache缓存

RDD Persist缓存

Cache和Persist有什么区别？

Persist的缓存级别

RDD CheckPoint检查点

cache与checkpoint的区别:

Shuffle会自动进行缓存

最后

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读