3.5 RDD持久化机制

news/2024/11/24 9:22:18/

一、RDD持久化

1、不采用持久化操作
查看要操作的HDFS文件
在这里插入图片描述
以集群模式启动Spark Shell

按照图示进行操作，得RDD4和RDD5

查看RDD4内容，会从RDD1到RDD2到RDD3到RDD4跑一趟

显示RDD5内容，也会从RDD1到RDD2到RDD3到RDD5跑一趟

在这里插入图片描述
2、采用持久化操作

计算RDD4，就是基于RDD3缓存的数据开始计算，不用从头到尾跑一趟

计算RDD5，就是基于RDD3缓存的数据开始计算，不用从头到尾跑一趟

二、存储级别

案例演示设置存储级别
在这里插入图片描述

package net.cxf.rdd.day05import org.apache.log4j.{Level, Logger}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}import java.awt.SystemTray/*** 功能：* 作者：cxf* 日期：2023年05月06日*/
object SetStorageLevel {def main(args: Array[String]): Unit = {//创建Spark配置对象val conf = new SparkConf().setAppName("SetStorageLevel") //设置应用名称.setMaster("local[*]") //设置主节点位置（本地调试>// 基于Spark配置对象创建Spark容器val sc = new SparkContext(conf)//去除Spark运行信息Logger.getLogger("org").setLevel(Level.OFF)Logger.getLogger("com").setLevel(Level.OFF)System.setProperty("spark.ui.showConsoleProgress", "false")Logger.getRootLogger().setLevel(Level.OFF)//创建RDDval rdd = sc.textFile("hdfs://master:9000/park/words.txt")//将rdd标记为持久化，采用默认存储级别- StorageLevel.MEMORY_ONLYrdd.persist() //无参持久化方法//对rdd做扁平映射,得到rdd1val rdd1 = rdd.flatMap(_.split(" "))//将rdd1持久化都磁盘rdd1.persist(StorageLevel.DISK_ONLY)//将rdd1映射成二元组，得到rdd2val rdd2 = rdd1.map((_, 1))//将rdd2持久化到内存，溢出的数据持久化到磁盘rdd2.persist(StorageLevel.MEMORY_AND_DISK)//第一次行动算子，对标记为持久化的RDD进行不同级别的持久化曹println("元素个数：" + rdd2.count)//第二次行动算子，直接利用rdd2的持久化数据进行操作，无须从头进行计算rdd2.collect.foreach(println)}}

运行程序，查看结果

在这里插入图片描述

三、利用Spark WebUI查看缓存

最好重启Spark Shell
在这里插入图片描述
（一）创建RDD并标记为持久化

（二）Spark WebUI查看RDD存储信息

收集RDD数据

刷新WebUI，发现出现了一个ParallelCollectionRDD的存储信息，该RDD的存储级别为MEMORY，持久化的分区为8，完全存储于内存中。

单击ParallelCollectionRDD超链接，可以查看该RDD的详细存储信息
在这里插入图片描述
执行以下命令，创建rdd2，并将rdd2持久化到磁盘

刷新上述WebUI，发现多了一个MapPartitionsRDD的存储信息，该RDD的存储级别为DISK，持久化的分区为8，完全存储于磁盘中。

（三）将RDD从缓存中删除

刷新上述WebUI，发现只剩下了MapPartitionsRDD，ParallelCollectionRDD已被移除。
在这里插入图片描述