一、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.day05
import 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)
    //创建RDD
    val rdd = sc.textFile("hdfs://master:9000/park/words.txt")
    //将rdd标记为持久化,采用默认存储级别- StorageLevel.MEMORY_ONLY
    rdd.persist() //无参持久化方法
    //对rdd做扁平映射,得到rdd1
    val rdd1 = rdd.flatMap(_.split(" "))
    //将rdd1持久化都磁盘
    rdd1.persist(StorageLevel.DISK_ONLY)
    //将rdd1映射成二元组,得到rdd2
    val 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已被移除。
 



















