详解 Spark 核心编程之 RDD 持久化

一、问题引出

/**
案例：对同一份数据文件分别做 WordCount 聚合操作和 Word 分组操作
期望：针对数据文件只进行一次分词、转换操作得到 RDD 对象，然后再对该对象分别进行聚合和分组，实现数据重用
*/
object TestRDDPersist {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[*]").setAppName("persist")
    	val sc = new SparkContext(conf)
        
        val rdd = sc.makeRDD(List(
        	"hello world", "hello spark"
        ))
        
        val flatRdd = rdd.flatMap(_.split(" "))
        
        val mapRdd = flatRdd.map(word => {
            println("@@@@@@@@@@")
            (word, 1)
        })
        
        // 聚合操作
        val reduceRdd = mapRdd.reduceByKey(_ + _)
        reduceRdd.collect().foreach(println)
        
        println("**********")
        
        // 分组操作
        val groupRdd = mapRdd.groupByKey()
        groupRdd.collect().foreach(println)
    }
}

/**
结果：flatRdd.map过程在聚合时和分组时分别都执行了，说明针对数据文件的分词、转换操作被重复执行了，只有对象被重用，而数据没有被重用
解析：
	1.RDD是不会存储数据的，当某个 RDD 转换成新的 RDD 后，该 RDD 中的数据就没有了
	2.如果需要再次用到该 RDD 的数据，则需要从数据源开始重新执行到该 RDD 来获取数据
解决：针对某个需要被重复使用的 RDD 对象在其进行下一步操作时先将数据进行缓存持久化或checkpoint，后续的其它操作从缓存持久化或checkpoint中获取数据
*/

二、RDD Cache

/**
缓存或持久化方法：
	1.rdd.cache()：底层调用 persist() 方法，默认是将数据保存到 JVM 堆内存中
	2.rdd.persist(StorageLevel)：可以指定数据的保存级别
说明：
	1.持久化方法被调用时不会立即进行缓存，而是在触发action算子时，数据才会被缓存在计算节点的内存中
	2.缓存除了用于数据重用，还可以提高容错性
*/
object TestRDDPersist {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[*]").setAppName("persist")
    	val sc = new SparkContext(conf)
        
        val rdd = sc.makeRDD(List(
        	"hello world", "hello spark"
        ))
        
        val flatRdd = rdd.flatMap(_.split(" "))
        
        val mapRdd = flatRdd.map(word => {
            println("@@@@@@@@@@")
            (word, 1)
        })
        
        //mapRdd.cache()
        mapRdd.persist()
        
        // 聚合操作
        val reduceRdd = mapRdd.reduceByKey(_ + _)
        reduceRdd.collect().foreach(println)
        
        println("**********")
        
        // 分组操作
        val groupRdd = mapRdd.groupByKey()
        groupRdd.collect().foreach(println)
        
        /*
        结果：聚合和分组前的操作过程只执行了一遍，实现了数据重用
        */
    }
}

// 存储级别
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)
}

三、RDD CheckPoint

/**
方法：rdd.checkpoint()，将 RDD 中间结果写入磁盘
说明：
	1.对 RDD 进行 checkpoint 操作并不会马上被执行，必须执行 Action 操作才能触发
	2.checkpoint保存由于在job执行完不会被删除，所以必须指定保存路径，一般保存在分布式文件系统
*/
object TestRDDPersist {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[*]").setAppName("persist")
    	val sc = new SparkContext(conf)
        
        // 指定checkpoint保存路径
        sc.setCheckpointDir("checkpoint")
        
        val rdd = sc.makeRDD(List(
        	"hello world", "hello spark"
        ))
        
        val flatRdd = rdd.flatMap(_.split(" "))
        
        val mapRdd = flatRdd.map(word => {
            println("@@@@@@@@@@")
            (word, 1)
        })
        
        mapRdd.checkpoint()
        
        // 聚合操作
        val reduceRdd = mapRdd.reduceByKey(_ + _)
        reduceRdd.collect().foreach(println)
        
        println("**********")
        
        // 分组操作
        val groupRdd = mapRdd.groupByKey()
        groupRdd.collect().foreach(println)
        
        /*
        结果：聚合和分组前的操作过程只执行了一遍，实现了数据重用
        */
    }
}

四、缓存和检查点区别

• cache 和 persist 会在原有的血缘关系中添加新的依赖，一旦数据出错可以重头读取数据；checkpoint 检查点会切断原有的血缘关系，重新建立新的血缘关系，相当于改变数据源
• cache 是将数据临时存储在 JVM 堆内存中，性能较高，但安全性低，persist 可以指定存储级别，将数据临时存储在磁盘文件中，涉及到 IO，性能较低，作业执行完毕后临时文件会被删除；checkpoint 是将数据长久地存储分布式文件系统中，安全性较高，但涉及 IO 且会独立开启一个作业从数据源开始获取数据，所以性能较低，一般在 checkpoint 前先进行 cache，当 checkpoint 时 job 只需从缓存中读取数据即可，可以提高性能