Spark的cache和persist

系统管理员 2022-07-13 14:45 156阅读 0赞

Spark的cache和persist

\------------------------------------------------------------------------------------------------------

cache调用的就是persist方法，默认存储级别只存储在内存中。

cache和persist的区别：cache只有一个默认的缓存级别MEMORY\_ONLY ，而persist可以根据情况设置其它的缓存级别。

![Center][]

\------------------------------------------------------------------------------------------------------

![Image 1][]

cache是Transformation类型的算子，只有在第一次计算使用之前才会被缓存。

![Center 1][]

\------------------------------------------------------------------------------------------------------

![Image 1][]

几种存储级别如下：

![Center 2][]

![Image 1][]

MEMORY\_ONLY

默认选项，RDD的（分区）数据直接以Java对象的形式存储于JVM的内存中，如果内存空间不足，某些分区的数据将不会被缓存，需要在使用的时候根据血统重新计算。

MYMORY\_AND\_DISK

RDD的数据直接以Java对象的形式存储于JVM的内存中，如果内存空间不中，某些分区的数据会被存储至磁盘，使用的时候从磁盘读取。

MEMORY\_ONLY\_SER

RDD的数据（Java对象）序列化之后存储于JVM的内存中（一个分区的数据为内存中的一个字节数组），相比于MEMORY\_ONLY能够有效节约内存空间（特别是使用一个快速序列化工具的情况下），但读取数据时需要更多的CPU开销；如果内存空间不足，处理方式与MEMORY\_ONLY相同。

MEMORY\_AND\_DISK\_SER

相比于MEMORY\_ONLY\_SER，在内存空间不足的情况下，将序列化之后的数据存储于磁盘。

DISK\_ONLY

仅仅使用磁盘存储RDD的数据（未经序列化）。

MEMORY\_ONLY\_2,MEMORY\_AND\_DISK\_2,

以MEMORY\_ONLY\_2为例，MEMORY\_ONLY\_2相比于MEMORY\_ONLY存储数据的方式是相同的，不同的是会将数据备份到集群中两个不同的节点，其余情况类似。

OFF\_HEAP(experimental)

RDD的数据序例化之后存储至Tachyon。相比于MEMORY\_ONLY\_SER，OFF\_HEAP能够减少垃圾回收开销、使得Spark Executor更“小”更“轻”的同时可以共享内存；

而且数据存储于Tachyon中，Spark集群节点故障并不会造成数据丢失，因此这种方式在“大”内存或多并发应用的场景下是很有吸引力的。需要注意的是，

Tachyon并不直接包含于Spark的体系之内，需要选择合适的版本进行部署；它的数据是以“块”为单位进行管理的，这些块可以根据一定的算法被丢弃，且不会被重建。

\------------------------------------------------------------------------------------------------------

一个例子：

import org.apache.spark.storage.StorageLevel
    import org.apache.spark.{SparkConf, SparkContext}
    /**
      * Created by User on 2016/11/2.
      */
    object cacheAndPersist {
      def main(args: Array[String]): Unit = {
        if(args.length!=1){
          System.err.println("Usage <Input>")
          System.exit(1)
        }
        val Array(input)=args
        //val input="E://sparkData/cacheAndPersist.txt"
        val conf: SparkConf = new SparkConf()
          .setMaster("local[2]")
          .setAppName("cacheAndPersist")
        val sc: SparkContext = new SparkContext(conf)
    
        val data1=sc.textFile(input)
            .map(_.split("\\|",100))
            .map(line=>{
              val Array(name,age)=line
              (name,age)
            }).cache()
        val data2=sc.textFile(input)
          .map(line => {line.split("\\|",100)})
          .map(x=>{
            val Array(name,age)=x
            (name,age)
          }).filter(y=>{y._1.equals("ML")}).persist(StorageLevel.MEMORY_AND_DISK)
    
        val data3=data1.intersection(data2).foreach(println)
      }
    }

输入：

![Center 3][]

运行结果：

![Image 1][]

![Center 4][]

[Center]: /images/20220713/b5437e7d1c4849fdbd5b878a7fedbcfc.png
[Image 1]: 
[Center 1]: /images/20220713/c8f6493a382d4a7c92b04aea05709003.png
[Center 2]: /images/20220713/810f3741280b458cb7dff60bd1be4a82.png
[Center 3]: /images/20220713/c1a479e12afa4954b8be8c0463a2bbe1.png
[Center 4]: /images/20220713/02e1a524f0294ef1a6a2da6dda6ac495.png