RDD的持久化-----rdd.cache()

叁歲伎倆 2022-11-20 10:55 248阅读 0赞

# 持久化 #

在Spark中，RDD采用惰性求值的机制，每次遇到行动操作，都会从头开始执行计算。每次调用行动操作，都会触发一次从头开始的计算，这对于迭代计算而言，代价是很大的，因为迭代计算经常需要多次重复使用同一组数据。下面就是多次计算同一个RDD的例子：

实际上，可以通过持久化（缓存）机制来避免这种重复计算的开销。具体方法是使用persist()方法对一个RDD标记为持久化，之所以说“标记持久化”，是因为出现persist()语句的地方，并不会马上计算生成RDD并把它持久化，而是要等到第一个行动操作触发真正计算以后，才会把计算结果进行持久化，持久化后的RDD将会被保留在计算节点的内存中，被后面的行动操作重复使用。persist()的圆括号中包含的是持久化级别参数，可以有如下不同的级别：

*  persist(MEMORY\_ONLY):表示将RDD作为反序列化的对象存储于JVM中，如果内存不足，就要按照LRU原则替换缓存中的内容。
 *  persist(MEMORY\_AND\_DISK):表示将RDD作为反序列化的对象存储在JVM中，如果内存不足，超出的分区将会被放在磁盘上。

一般而言，使用cache()方法时，会调用persist(MEMORY\_ONLY)。针对上面的实例，增加持久化语句的执行过程如下：

scala>  val list=List("Hadoop","Spark","Hive")
    list: List[String] = List(Hadoop, Spark, Hive)
    
    scala> val rdd=sc.parallelize(list)
    rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[13] at parallelize at <console>:26
    
    scala> rdd.cache()//会调用persist(MEMORY_ONLY)，但是，语句执行到这里，并不会缓存rdd，因为这时rdd还没被计算生成
    res15: rdd.type = ParallelCollectionRDD[13] at parallelize at <console>:26
    
    scala> println(rdd.count())//第一次行动操作，触发一次真正从头到尾的计算，这时上面的rdd.cache()才会被执行，把这个rdd放到缓存中
    3
    
    scala> println(rdd.collect().mkString(","))//第二次行动操作，不需要触发从头到尾的计算，只需要重复使用上面缓存中的rdd
    Hadoop,Spark,Hive

持久化RDD会占用内存空间，当不再需要一个RDD时，就可以使用unpersist()方法手动地把持久化RDD从缓存中移除，释放内存空间。

rdd.cache()//表示对rdd进行持久化，把它保存在内存或磁盘中（这里采用cache方法把数据集保存在内存中）
    方便后续重复使用，当数据被反复访问时（比如查询一些热点数据，或者运行迭代算法），这是非常有用的，而且
    通过cache()可以缓存非常大的数据集，支持跨越几十甚至上百个节点。

补充一些collect的知识

rdd.collect()//collect()方法会把各个worker上的所有RDD元素都抓取到Driver中，
    因此，可能会导致driver所在节点发生内存溢出。
    因此，当只需要打印RDD的部分元素时，可以采用类似于rdd.take(100).foreach(println)

PS：内容来自《Spark编程基础》这本书，记录的原因是我对这部分知识掌握得不太好，想借助博客把不太懂的部分记录下来

侵删。如果感兴趣，可以去厦门大学数据库实验室的官网学习。