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

news2025/1/20 18:27:14

循循渐进理解

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没有缓存，所以要执行两次
在这里插入图片描述

上述的问题
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条
在这里插入图片描述

发现有个绿色点
在这里插入图片描述
发现cache存到memory里面

RDD的持久化分为
缓存

数据保存位置: task所在主机内存/本地磁盘中
数据保存时机: 在缓存所在第一个Job执行过程中进行数据保存
使用: rdd.cache()/rdd.persist()/rdd.persist(StorageLevel.XXXX)
cache与persist的区别
- cache是只将数据保存在内存中(cache的底层就是persisit())
- persist是可以指定将数据保存在内存/磁盘中
常用的存储级别:
- 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)
  }

}