Spark Core快速入门系列(9) | RDD缓存和设置检查点

不温卜火

发布于 2020-10-28 16:58:06

7690

发布于 2020-10-28 16:58:06

文章被收录于专栏：不温卜火

一. RDD缓存

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

通过查看源码发现cache最终也是调用了persist方法，默认的存储级别都是仅在内存存储一份，Spark的存储级别还有好多种，存储级别在object StorageLevel中定义的。

在存储级别的末尾加上“_2”来把持久化数据存为两份

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

// 1.创建一个RDD
scala> val rdd = sc.makeRDD(Array("buwenbuhuo"))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[19] at makeRDD at <console>:25

// 2.将RDD转换为携带当前时间戳不做缓存
scala> val nocache = rdd.map(_.toString+System.currentTimeMillis)
nocache: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[20] at map at <console>:27

// 3.多次打印结果
scala> nocache.collect
res0: Array[String] = Array(buwenbuhuo1538978275359)

scala> nocache.collect
res1: Array[String] = Array(buwenbuhuo1538978282416)

scala> nocache.collect
res2: Array[String] = Array(buwenbuhuo1538978283199)

// 4.将RDD转换为携带当前时间戳并做缓存
scala> val cache =  rdd.map(_.toString+System.currentTimeMillis).cache
cache: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[21] at map at <console>:27

// 5.多次打印做了缓存的结果
scala> cache.collect
res3: Array[String] = Array(buwenbuhuo1538978435705)                                   

scala> cache.collect
res4: Array[String] = Array(buwenbuhuo1538978435705)

scala> cache.collect
res5: Array[String] = Array(buwenbuhuo1538978435705)

二. 设置检查点（checkpoint）

Spark 中对于数据的保存除了持久化操作之外，还提供了一种检查点的机制,检查点（本质是通过将RDD写入Disk做检查点）是为了通过 Lineage 做容错的辅助

Lineage 过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的 RDD 开始重做 Lineage，就会减少开销。

检查点通过将数据写入到 HDFS 文件系统实现了 RDD 的检查点功能。

为当前 RDD 设置检查点。该函数将会创建一个二进制的文件，并存储到 checkpoint 目录中，该目录是用 SparkContext.setCheckpointDir()设置的。在 checkpoint 的过程中，该RDD 的所有依赖于父 RDD中的信息将全部被移除。

对 RDD 进行 checkpoint 操作并不会马上被执行，必须执行 Action 操作才能触发, 在触发的时候需要对这个 RDD 重新计算.

1. 代码

package Day04
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 **
@author 不温卜火
 **
 * @create 2020-07-26 15:35
 **
 *         MyCSDN :https://buwenbuhuo.blog.csdn.net/
 */
object CheckPointDemo {
  def main(args: Array[String]): Unit = {
    // 要在SparkContext初始化之前设置, 都在无效
    System.setProperty("HADOOP_USER_NAME", "buwenbuhuo")
    val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
    val sc = new SparkContext(conf)
    // 设置 checkpoint的目录. 如果spark运行在集群上, 则必须是 hdfs 目录
    sc.setCheckpointDir("./ck1")
    val rdd1 = sc.parallelize(Array("abc"))
    val rdd2: RDD[String] = rdd1.map(_ + " : " + System.currentTimeMillis())

    /*
    标记 RDD2的 checkpoint.
    RDD2会被保存到文件中(文件位于前面设置的目录中), 并且会切断到父RDD的引用, 也就是切断了它向上的血缘关系
    该函数必须在job被执行之前调用.
    强烈建议把这个RDD序列化到内存中, 否则, 把他保存到文件的时候需要重新计算.
     */
    rdd2.checkpoint()
    rdd2.collect().foreach(println)
    rdd2.collect().foreach(println)
    rdd2.collect().foreach(println)
  }
}

2. 结果

3. 持久化和checkpoint的区别

持久化只是将数据保存在 BlockManager 中，而 RDD 的 Lineage 是不变的。但是checkpoint 执行完后，RDD 已经没有之前所谓的依赖 RDD 了，而只有一个强行为其设置的checkpointRDD，RDD 的 Lineage 改变了。
持久化的数据丢失可能性更大，磁盘、内存都可能会存在数据丢失的情况。但是 checkpoint 的数据通常是存储在如 HDFS 等容错、高可用的文件系统，数据丢失可能性较小。
注意: 默认情况下，如果某个 RDD 没有持久化，但是设置了checkpoint，会存在问题. 本来这个 job 都执行结束了，但是由于中间 RDD 没有持久化，checkpoint job 想要将 RDD 的数据写入外部文件系统的话，需要全部重新计算一次，再将计算出来的 RDD 数据 checkpoint到外部文件系统。所以，建议对 checkpoint()的 RDD 使用持久化, 这样 RDD 只需要计算一次就可以了.

本次的分享就到这里了

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2020/07/30 ，如有侵权请联系 cloudcommunity@tencent.com 删除

存储