返回 导航

Spark

hangge.com

Spark - RDD使用详解8(持久化:缓存、检查点)

作者:hangge | 2023-10-18 09:21

十、RDD 缓存(Cache)

1,缓存的使用

(1)缓存是指将 RDD 的计算结果存储在内存中,以便后续的重复使用。通过缓存 RDD,可以避免每次需要使用 RDD 时都重新计算,从而提高 Spark 应用程序的性能。
  • 比如下面样例,我们分别实现了单词数量统计和单词分组这两个功能,虽然 reduceRDDgroupRDD 都是从 mapRDD 转换而来,但是每次行动前面的所有操作都会从头执行一次。
// 创建 Spark 运行配置对象
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

// 创建 Spark 上下文环境对象(连接对象)
val sc: SparkContext = new SparkContext(sparkConf)

// 创建RDD
val listRDD = sc.makeRDD(List("welcome to hangge.com", "hello world", "hangge.com"))

// 对RDD进行转换操作
val wordRDD: RDD[String] = listRDD.flatMap(_.split(" "))
val mapRDD: RDD[(String, Int)] = wordRDD.map {
  word => {
    println("*** wordRDD.map ***")
    (word, 1)
  }
}

// 统计单词数量
val reduceRDD = mapRDD.reduceByKey(_+_)
reduceRDD.collect().foreach(println)
println("-------------------------")

// 单词分组
val groupRDD = mapRDD.groupByKey()
groupRDD.collect().foreach(println)

//关闭 Spark
sc.stop()

(2)RDD 通过 Cache 或者 Persist 方法将前面的计算结果缓存,默认情况下会把数据以缓存在 JVM 的堆内存中。但是并不是这两个方法被调用时立即缓存,而是触发后面的 action 算子时,该 RDD 将会被缓存在计算节点的内存中,并供后面重用。
  • 下面样例我们将 mapRDD 缓存到内存中,这样后面再次使用时就不需要重新计算了:
// 创建 Spark 运行配置对象
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

// 创建 Spark 上下文环境对象(连接对象)
val sc: SparkContext = new SparkContext(sparkConf)

// 创建RDD
val listRDD = sc.makeRDD(List("welcome to hangge.com", "hello world", "hangge.com"))

// 对RDD进行转换操作
val wordRDD: RDD[String] = listRDD.flatMap(_.split(" "))
val mapRDD: RDD[(String, Int)] = wordRDD.map {
  word => {
    println("*** wordRDD.map ***")
    (word, 1)
  }
}

// 将RDD缓存到内存中
mapRDD.cache()

// 统计单词数量
val reduceRDD = mapRDD.reduceByKey(_+_)
reduceRDD.collect().foreach(println)
println("-------------------------")

// 单词分组
val groupRDD = mapRDD.groupByKey()
groupRDD.collect().foreach(println)

//关闭 Spark
sc.stop()

2,缓存级别

(1)Spark 提供了不同的缓存级别,可以根据需求选择适合的级别。常见的缓存级别包括:
  • MEMORY_ONLY:将 RDD 存储在内存中,但不序列化。如果内存不足,可能会丢失数据。
  • MEMORY_AND_DISK:将 RDD 存储在内存中,如果内存不足,则溢出到磁盘。
  • MEMORY_ONLY_SER:将 RDD 存储在内存中,并使用序列化方式。
  • MEMORY_AND_DISK_SER:将 RDD 存储在内存中,并使用序列化方式,如果内存不足,则溢出到磁盘。
  • OFF_HEAP:将 RDD 存储在堆外内存中,通过 DirectByteBuffer 实现,适用于大规模数据集。

(2)使用 persist() 方法可以指定存储级别进行缓存:
注意:
  • 默认的存储级别都是仅在内存存储一份。在存储级别的末尾加上“2”表示持久化的数据存为两份。
  • cache() 底层其实就是调用了 persist(),并设置缓存级别 MEMORY_ONLY
mapRDD.persist(StorageLevel.MEMORY_AND_DISK_2)

3,缓存的一些注意事项

(1)缓存有可能丢失,或者存储于内存的数据由于内存不足而被删除,RDD 的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于 RDD 的一系列转换,丢失的数据会被重算,由于 RDD 的各个 Partition 是相对独立的,因此只需要计算丢失的部分即可,并不需要重算全部 Partition
(2)Spark 会自动对一些 Shuffle 操作的中间数据做持久化操作(比如:reduceByKey)。这样做的目的是为了当一个节点 Shuffle 失败了避免重新计算整个输入。但是,在实际使用的时候,如果想重用数据,仍然建议调用 persistcache

十一、RDD 检查点(CheckPoint)

1,检查点介绍

    所谓的检查点其实就是通过将 RDD 中间结果写入磁盘。由于血缘依赖过长会造成容错成本过高,这样就不如在中间阶段做检查点容错,如果检查点之后有节点出现问题或应用程序重新计算时,可以从检查点开始重做血缘,减少了开销。

2,检查点的使用

(1)设置检查点分为如下两个步骤:
  • 指定一个检查点目录,用于存储检查点 RDD 的数据。这个目录可以是本地文件系统的路径,也可以是分布式文件系统(如 HDFS)的路径。
  • 使用 checkpoint() 方法将 RDD 设置为检查点。当该方法调用时,Spark 将计算 RDD 的所有转换和操作,并将计算结果写入持久化存储。这会创建一个具有计算结果的新的 RDD,以后的操作将使用该检查点 RDD
// 创建 Spark 运行配置对象
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
// 创建 Spark 上下文环境对象(连接对象)
val sc: SparkContext = new SparkContext(sparkConf)

// 设置检查点目录(如HDFS)
sc.setCheckpointDir("./checkpoint1")

// 创建RDD
val listRDD = sc.makeRDD(List("welcome to hangge.com", "hello world", "hangge.com"))

// 对RDD进行转换操作
val wordRDD: RDD[String] = listRDD.flatMap(_.split(" "))
val mapRDD: RDD[(String, Int)] = wordRDD.map {
  word => {
    println("*** wordRDD.map ***")
    (word, 1)
  }
}

// 设置检查点
mapRDD.checkpoint()

// 统计单词数量
val reduceRDD = mapRDD.reduceByKey(_+_)
reduceRDD.collect().foreach(println)
println("-------------------------")

// 单词分组
val groupRDD = mapRDD.groupByKey()
groupRDD.collect().foreach(println)

//关闭 Spark
sc.stop()

(2)执行后可以看到设置的检查点目录下会生成如下一些二进制文件:

(3)控制台输出内容如下:
注意:可以看到到 mapRDD 的转换计算其实执行了两次,其中第二次并不是由于复用造成的,而是因为将 RDD 设置为检查点时,该 RDD 的计算将会被强制执行并写入到持久化存储中。因此,检查点虽然可以避免重复计算,但也会引入额外的存储和计算开销。我们需要在性能和资源利用之间进行权衡,并选择合适的 RDD 进行检查点操作。

附:缓存和检查点区别

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

全部评论(0)

回到顶部