4、RDD持久化
4.1 RDD Cache缓存
1、RDD Cache缓存
（1）RDD通过Cache或者persist方法将前面的计算结果缓存
（2）默认情况下会把数据以序列化的形式缓存在JVM的堆内存中。
（3）但是并不是这个两个方法被调用时立即缓存，而是触发后面的action算子时，该RDD将会被缓存在计算节点的内存中，并供后面重用。
2、创建包名com.zhm.spark.operator.cache
3、未使用缓存代码实现
package com.zhm.spark.operator.cache; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @ClassName Test01_no_cache * @Description TODO * @Author Zouhuiming * @Date 2023/6/29 13:58 * @Version 1.0 */ public class Test01_no_cache { public static void main(String[] args) throws InterruptedException { //1、创建配置对象 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Test01"); //2、创建sparkContext JavaSparkContext sparkContext = new JavaSparkContext(conf); //3、创建RDD数据集 JavaRDD<String> sourceRDD = sparkContext.textFile("./input/word.txt"); //4、炸裂每行数据 JavaRDD<String> flatMapRDD = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.stream(s.split(" ")).iterator(); } }); //5、转换元素形式-->(word,1) JavaPairRDD<String, Integer> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { System.out.println("你看我出现几次？"); return new Tuple2<>(s, 1); } }); //6、执行两个Action算子，以触发两个Job mapToPairRDD.saveAsTextFile("output/1"+System.currentTimeMillis()+".txt"); mapToPairRDD.saveAsTextFile("output/2"+System.currentTimeMillis()+".txt"); //7、设置睡眠时间 以便查看localhost:4040 Thread.sleep(1000000L); //x 关闭 sparkContext sparkContext.stop(); } }
4、运行结果：
结论：一共输出12次，我们的RDD中有12条数据，每个job都执行一个flatMapRDD.map都会输出，所以是8次，也就意味着我们的两个job都会从头开始计算，直到最终的结果。
查看任务的WebUI
5、使用缓存代码实现
package com.zhm.spark.operator.cache; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @ClassName Test02_has_cache * @Description TODO * @Author Zouhuiming * @Date 2023/6/29 14:08 * @Version 1.0 */ public class Test02_has_cache { /** *自带缓存的算子：reduceByKey */ public static void main(String[] args) throws InterruptedException { //1、创建配置对象 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Test01"); //2、创建sparkContext JavaSparkContext sparkContext = new JavaSparkContext(conf); //3、创建RDD数据集 JavaRDD<String> sourceRDD = sparkContext.textFile("./input/word.txt"); //4、炸裂每行数据 JavaRDD<String> flatMapRDD = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.stream(s.split(" ")).iterator(); } }); //5、转换元素形式-->(word,1) JavaPairRDD<String, Integer> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { System.out.println("你看我出现几次？"); return new Tuple2<>(s, 1); } }); //将mapToPairRDD进行缓存 mapToPairRDD.cache(); //6、执行两个Action算子，以触发两个Job mapToPairRDD.saveAsTextFile("output/1"+System.currentTimeMillis()+".txt"); mapToPairRDD.saveAsTextFile("output/2"+System.currentTimeMillis()+".txt"); //7、设置睡眠时间 以便查看localhost:4040 Thread.sleep(1000000L); //x 关闭 sparkContext sparkContext.stop(); } }
注意：缓存的应用程序执行结束之后，缓存的目录也会被删除
6、运行结果
结论：我们对JavaPairRDD进行了缓存，那么也就第一个Job会从头到JavaPairRDD执行，而第二个则会从缓存中得到JavaPairRDD数据，继续自己的处理逻辑。
观察任务WebUI
不同于未使用缓存的任务，本次job中再map出对javaPairRDD进行了缓存，途中绿色的点就表示缓存，鼠标停留到绿色点处会有提示。
7、缓存相关源码解析
mapRdd.cache() def cache(): this.type = persist() def persist(): this.type = persist(StorageLevel.MEMORY_ONLY) 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)
注意：默认的存储级别都是仅在内存存储一份。在存储级别的末尾加上“_2”表示持久化的数据存为两份。SER：表示序列化。
8、SparkRDD的安全问题和解决方案
（1）问题：缓存有可能会丢失，或者储存于内存的数据由于内存不足而被删除
（2）RDD的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行
（3）原理：
通过基于RDD的一系列转换，丢失的数据会被重算，由于RDD的各个Partition是相对独立的，因此只需要计算丢失的部分即可，并不需要重算所有的Partition。
9、自带缓存算子
（1）Spark会自动对一些Shuffle操作的中间数据做持久化操作（别人ReduceByKey）。
（2）这样做的目的是为了当一个节点Shuffle失败了避免重新计算整个输入。
（3）但是，在时间使用的时候，如果想重用数据，仍然建议调用persist或Cache。
（4）编写具有自带缓存算子的代码
package com.zhm.spark.operator.cache; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @ClassName Test03_OperatorWithCache * @Description TODO * @Author Zouhuiming * @Date 2023/6/29 14:19 * @Version 1.0 */ public class Test03_OperatorWithCache { public static void main(String[] args) throws InterruptedException { //1、创建配置对象 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Test01"); //2、创建sparkContext JavaSparkContext sparkContext = new JavaSparkContext(conf); //3、创建RDD数据集 JavaRDD<String> sourceRDD = sparkContext.textFile("input/word.txt"); //4、炸裂每行数据 JavaRDD<String> flatMapRDD = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.stream(s.split(" ")).iterator(); } }); //5、转换元素形式-->(word,1) JavaPairRDD<String, Integer> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { System.out.println("你看我出现几次？"); return new Tuple2<>(s, 1); } }); //6、对mapToPairRDD按照key聚合 JavaPairRDD<String, Integer> reduceByKeyRDD = mapToPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1 + v2; } }); //7、执行两个行动算子 reduceByKeyRDD.saveAsTextFile("output/1"+System.currentTimeMillis()+".txt"); reduceByKeyRDD.saveAsTextFile("output/2"+System.currentTimeMillis()+".txt"); //8、设置睡眠时间 以便查看localhost:4040 Thread.sleep(1000000L); //x 关闭 sparkContext sparkContext.stop(); } }
访问http://localhost:4040/jobs/页面，查看第一个和第二个job的DAG图。
4.2 RDD CheckPoint检查点
1、检查点：是将RDD中间结果写入磁盘
2、为什么要做检查点？
由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以就从检查点开始重做血缘，减少了开销。
3、检查点储存路径：Checkpoint的数据通常是储存在HDFS等容错、高可用的文件系统。
4、检查点数据储存格式为：二进制文件
5、检查点切断血缘：在Checkpoint的过程中，该RDD的所以依赖于父RDD的信息将全部被移除。
6、检查点触发时间：对RDD进行Checkpoint操作并不会马上执行，必须执行Action操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍
7、设置检查点步骤
（1）设置检查点数据储存路径：sc.setCheckpointDir(“./checkpoint1”)
（2）调用检查点方法：wordToOneRdd.checkpoint()
8、代码实现
package com.zhm.spark.operator.cache; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @ClassName Test04_checkPoint * @Description TODO * @Author Zouhuiming * @Date 2023/6/29 14:26 * @Version 1.0 */ public class Test04_checkPoint { public static void main(String[] args) { //1、创建配置对象 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Test01"); //2、创建sparkContext JavaSparkContext sparkContext = new JavaSparkContext(conf); //设置检查点储存路径 sparkContext.setCheckpointDir("output/checkPoint"); //3、创建RDD数据集 JavaRDD<String> sourceRDD = sparkContext.textFile("input/word.txt"); //4、炸裂每行数据 JavaRDD<String> flatMapRDD = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.stream(s.split(" ")).iterator(); } }); //5、转换元素形式-->(word,1) JavaPairRDD<String, Integer> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { System.out.println("你看我出现几次？"); return new Tuple2<>(s, 1); } }); //6、对mapToPairRDD设置检查点 mapToPairRDD.checkpoint(); //7、将数据储存到文件中 mapToPairRDD.saveAsTextFile("output/checkPoint/wordCount_"+System.currentTimeMillis()+".txt"); //x 关闭 sparkContext sparkContext.stop(); } }
9、执行结果：
访问http://localhost:4040/jobs/
10、Checkpoint对血缘的一些
package com.zhm.spark.operator.cache; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @ClassName Test05_checkPoint_printLineage * @Description TODO * @Author Zouhuiming * @Date 2023/6/29 15:12 * @Version 1.0 */ public class Test05_checkPoint_printLineage { public static void main(String[] args) throws InterruptedException { //1、创建配置对象 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Test01"); //2、创建sparkContext JavaSparkContext sparkContext = new JavaSparkContext(conf); //设置检查点储存路径 sparkContext.setCheckpointDir("output/checkpoint1"); //3、创建RDD数据集 JavaRDD<String> sourceRDD = sparkContext.textFile("input/word.txt"); //4、炸裂每行数据 JavaRDD<String> flatMapRDD = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.stream(s.split(" ")).iterator(); } }); //5、转换元素形式-->(word,1) JavaPairRDD<String, Integer> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { System.out.println("你看我出现几次？"); return new Tuple2<>(s, 1); } }); //6、在检查点之前答应血缘 System.out.println("在检查点之前打印血缘: "); System.out.println(mapToPairRDD.toDebugString()); //7、对mapToPairRDD设置检查点 mapToPairRDD.checkpoint(); //8、将数据储存到文件中 mapToPairRDD.saveAsTextFile("output/checkPoint1/wordCount"+System.currentTimeMillis()+".txt"); //9、在检查点之后打印血缘 System.out.println("在检查点之后打印血缘: "); System.out.println(mapToPairRDD.toDebugString()); Thread.sleep(10000000); //x 关闭 sparkContext sparkContext.stop(); } }
运行结果：
结论：血缘关系被切断了，因为Checkpoint机制是储存的数据很安全了，不用保留血缘依赖。
11、Checkpoint对数据的影响
package com.zhm.spark.operator.cache; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @ClassName Test06_check_influence_data * @Description TODO * @Author Zouhuiming * @Date 2023/6/29 15:21 * @Version 1.0 */ public class Test06_check_influence_data { public static void main(String[] args) { //1、创建配置对象 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Test01"); //2、创建sparkContext JavaSparkContext sparkContext = new JavaSparkContext(conf); //设置检查点储存路径 sparkContext.setCheckpointDir("output/checkpoint3"); //3、创建RDD数据集 JavaRDD<String> sourceRDD = sparkContext.textFile("input/word.txt"); //4、炸裂每行数据 JavaRDD<String> flatMapRDD = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.stream(s.split(" ")).iterator(); } }); //5、转换元素形式-->(word,当前时间) JavaPairRDD<String, Long> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Long>() { @Override public Tuple2<String, Long> call(String s) throws Exception { System.out.println("你看我出现几次？"); return new Tuple2<>(s, System.currentTimeMillis()); } }); //6、checkpoint前收集打印 System.out.println("checkpoint前收集打印："); mapToPairRDD.collect().forEach(System.out::println); //7、对mapToPairRDD设置检查点 mapToPairRDD.checkpoint(); //6、checkpoint后收集打印 System.out.println("checkpoint前收集打印："); mapToPairRDD.collect().forEach(System.out::println); //x 关闭 sparkContext sparkContext.stop(); } }
运行结果：
现象：由于Checkpoint要从头再执行一遍，这种与时间相关的就会造成数据不一致。
12、Checkpoint检查点+Cache缓存
package com.zhm.spark.operator.cache; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @ClassName Test07_checkPoint_with_cache * @Description TODO * @Author Zouhuiming * @Date 2023/6/29 15:27 * @Version 1.0 */ public class Test07_checkPoint_with_cache { public static void main(String[] args) { //1、创建配置对象 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Test01"); //2、创建sparkContext JavaSparkContext sparkContext = new JavaSparkContext(conf); //设置检查点储存路径 sparkContext.setCheckpointDir("output/checkpoint4"); //3、创建RDD数据集 JavaRDD<String> sourceRDD = sparkContext.textFile("input/word.txt"); //4、炸裂每行数据 JavaRDD<String> flatMapRDD = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.stream(s.split(" ")).iterator(); } }); //5、转换元素形式-->(word,当前时间) JavaPairRDD<String, Long> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Long>() { @Override public Tuple2<String, Long> call(String s) throws Exception { System.out.println("你看我出现几次？"); return new Tuple2<>(s, System.currentTimeMillis()); } }); //缓存 mapToPairRDD.cache(); //6、checkpoint前收集打印 System.out.println("checkpoint前收集打印："); mapToPairRDD.collect().forEach(System.out::println); //7、对mapToPairRDD设置检查点 mapToPairRDD.checkpoint(); //6、checkpoint后收集打印 System.out.println("checkpoint后收集打印："); mapToPairRDD.collect().forEach(System.out::println); //x 关闭 sparkContext sparkContext.stop(); } }
运行结果：
结论：保持了数据的一致性
4.3缓存和检查点的区别
1、Cache缓存只是将数据保存起来，不切断血缘依赖.Checkpoint检查点切断血缘依赖
2、Cache缓存的数据通常储存在磁盘、内存等地方，可靠性低。Checkpoint的数据通常储存在HDFS等容错、高可用的文件系统，可靠性高。
3、建议对Checkpoint的RDD使用Cache缓存，这样Checkpoint的job只需要从Cache缓存中读取数据即可，否者需要再从头计算一次RDD。
4、如果使用完了缓存，可以通过unpersist方法是否缓存。
4.4. 检查点储存到HDFS集群
注意：如果检查点数据储存到HDFS集群，要注意配置访问集群的用户名。否者会报访问权限异常。
package com.zhm.spark.operator.cache; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @ClassName Test08_checkpoint_hdfs * @Description TODO * @Author Zouhuiming * @Date 2023/6/29 15:30 * @Version 1.0 */ public class Test08_checkpoint_hdfs { public static void main(String[] args) { //0、设置hadoop用户 System.setProperty("HADOOP_USER_NAME","zhm"); //1、创建配置对象 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("Test01"); //2、创建sparkContext JavaSparkContext sparkContext = new JavaSparkContext(conf); //设置检查点储存路径 sparkContext.setCheckpointDir("hdfs://hadoop102:8020/sparkCheckPoint"); //3、创建RDD数据集 JavaRDD<String> sourceRDD = sparkContext.textFile("input/word.txt"); //4、炸裂每行数据 JavaRDD<String> flatMapRDD = sourceRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.stream(s.split(" ")).iterator(); } }); //5、转换元素形式-->(word,1) JavaPairRDD<String, Integer> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { System.out.println("你看我出现几次？"); return new Tuple2<>(s, 1); } }); //6、对mapToPairRDD设置检查点 mapToPairRDD.checkpoint(); //7、将数据储存到文件中 mapToPairRDD.saveAsTextFile("output/checkPoint5/wordCount_"+System.currentTimeMillis()+".txt"); //x 关闭 sparkContext sparkContext.stop(); } }
