RDD持久化

﹏ヽ暗。殇╰゛Y 2022-09-06 15:23 156阅读 0赞

#### RDD持久化原理 ####

Spark中有一个非常重要的功能就是可以对RDD进行持久化。  
当对RDD执行持久化操作时，每个节点都会将自己操作的RDD的partition数据持久化到内存中，并且在之后对该RDD的反复使用中，直接使用内存中缓存的partition数据。

这样的话，针对一个RDD反复执行多个操作的场景，就只需要对RDD计算一次即可，后面直接使用该RDD，而不需要反复计算多次该RDD。  
因为正常情况下这个RDD的数据使用过后内存中是不会一直保存的。  
例如这样的操作:  
针对mapRDD需要多次使用的

val dataRDD = sc.parallelize(Array(1,2,3,4,5))
    val mapRDD = dataRDD.map(...)
    mapRDD.foreach(...)
    mapRDD.saveAsTextFile(...)
    mapRDD.collect()

巧妙使用RDD持久化，在某些场景下，对spark应用程序的性能有很大提升。  
特别是对于迭代式算法和快速交互式应用来说，RDD持久化，是非常重要的。

要持久化一个RDD，只需要调用它的`cache()`或者`persist()`方法就可以了。  
在该RDD第一次被计算出来时，就会直接缓存在每个节点中。而且Spark的持久化机制还是自动容错的，如果持久化的RDD的任何partition数据丢失了，那么Spark会自动通过其源RDD，使用transformation算子重新计算该partition的数据。

cache()和persist()的区别在于:  
cache()是persist()的一种简化方式，cache()的底层就是调用的persist()的无参版本，也就是调用persist(MEMORY\_ONLY)，将数据持久化到内存中。  
如果需要从内存中清除缓存，那么可以使用unpersist()方法。

#### RDD持久化策略 ####

下面看一下目前Spark支持的一些持久化策略:

<table> 
 <thead> 
  <tr> 
   <th>策略</th> 
   <th>介绍</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>MEMORY_ONLY</td> 
   <td>以非序列化的方式持久化在JVM内存中</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK</td> 
   <td>同上，但是当某些partition无法存储在内存中时，会持久化到磁盘中</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_SER</td> 
   <td>同MEMORY_ONLY，但是会序列化</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK_SER</td> 
   <td>同MEMORY_AND_DSK，但是会序列化</td> 
  </tr> 
  <tr> 
   <td>DISK_ONLY</td> 
   <td>以非序列化的方式完全存储到磁盘上</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_2、MEMORY_AND_DISK_2等</td> 
   <td>如果是尾部加了2的持久化级别，表示会将持久化数据复用一份，保存到其他节点</td> 
  </tr> 
 </tbody> 
</table>

**补充说明：**

*  MEMORY\_ONLY:以非序列化的Java对象的方式持久化在JVM内存中。如果内存无法完全存储RDD所有的partition，那么那些没有持久化的partition就会在下一次需要使用它的时候，重新被计算。
 *  MEMORY\_AND\_DISK:当某些partition无法存储在内存中时，会持久化到磁盘中。下次需要使用这些partition时，需要从磁盘上读取，不需要重新计算
 *  MEMORY\_ONLY\_SER:同MEMORY\_ONLY，但是会使用Java的序列化方式，将Java对象序列化后进行持久化。可以减少内存开销，但是在使用的时候需要进行反序列化，因此会增加CPU开销。
 *  MEMORY\_AND\_DISK\_SER:同MEMORY\_AND\_DSK。但是会使用序列化方式持久化Java对象。
 *  DISK\_ONLY:使用非序列化Java对象的方式持久化，完全存储到磁盘上。
 *  MEMORY\_ONLY\_2、MEMORY\_AND\_DISK\_2等:如果是尾部加了2的持久化级别，表示会将持久化数据复制一份，保存到其它节点，从而在数据丢失时，不需要重新计算，只需要使用备份数据即可。

#### 如何选择RDD持久化策略 ####

Spark提供了多种持久化级别，主要是为了在CPU和内存消耗之间进行取舍。

下面是一些通用的持久化级别的选择建议:

1.  **优先使用MEMORY\_ONLY**，纯内存速度最快，而且没有序列化不需要消耗CPU进行反序列化操作，缺点就是比较耗内存
2.  **MEMORY\_ONLY\_SER**，将数据进行序列化存储，纯内存操作还是非常快，只是在使用的时候需要消耗CPU进行反序列化

> 注意:  
> 如果需要进行数据的快速失败恢复，那么就选择带后缀为\_2的策略，进行数据的备份，这样在失败时，就不需要重新计算了  
> 能不使用DISK相关的策略，就不要使用，因为有的时候，从磁盘读取数据，还不如重新计算一次。

#### 案例:使用RDD的持久化 ####

下面来写一个案例验证一下RDD持久化的效果

在没有添加cache之前，每一次都耗时很长  
加上cache之后，第二次计算耗时就很少了

**Scala版本**

import org.apache.spark.{ SparkConf, SparkContext}
    
    /** * 需求：RDD持久化 */
    object PersistRddScala { 
    
      	def main(args: Array[String]): Unit = { 
    	    val conf = new SparkConf()
    	    conf.setAppName("PersistRddScala")
    	      .setMaster("local")
    	    val sc = new SparkContext(conf)
    	
    	    //注意cache的用法和位置
    	    //cache默认是基于内存的持久化
    	    // cache()=persist()=persist(StorageLevel.MEMORY_ONLY)
    	    val dataRDD = sc.textFile("~/hello_10000000.dat").cache()
    	    var start_time = System.currentTimeMillis()
    	    var count = dataRDD.count()
    	    println(count)
    	    var end_time = System.currentTimeMillis()
    	    println("第一次耗时："+(end_time-start_time))
    	
    	
    	    start_time = System.currentTimeMillis()
    	    count = dataRDD.count()
    	    println(count)
    	    end_time = System.currentTimeMillis()
    	    println("第二次耗时："+(end_time-start_time))
    	
    	    sc.stop()
      	}
    }

**Java版本**

import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    
    /** * 需求：RDD持久化 */
    public class PersistRddJava { 
    
        public static void main(String[] args) { 
            //创建JavaSparkContext
            SparkConf conf = new SparkConf();
            conf.setAppName("PersistRddJava")
                    .setMaster("local");
            JavaSparkContext sc = new JavaSparkContext(conf);
    
            JavaRDD<String> dataRDD = sc.textFile("~/hello_10000000.dat").cache();
    
            long start_time = System.currentTimeMillis();
            long count = dataRDD.count();
            System.out.println(count);
            long end_time = System.currentTimeMillis();
            System.out.println("第一次耗时："+(end_time-start_time));
    
    
            start_time = System.currentTimeMillis();
            count = dataRDD.count();
            System.out.println(count);
            end_time = System.currentTimeMillis();
            System.out.println("第二次耗时："+(end_time-start_time));
    
            sc.stop();
        }
    
    }