Spark 持久化(缓存)

短命女 2022-04-12 04:54 186阅读 0赞

# Spark 持久化(缓存)  #

如前所述，Spark RDD 是惰性求值的，而有时我们希望能多次使用同一个RDD。如果简单地对RDD 调用行动操作，Spark 每次都会重算RDD 以及它的所有依赖。这在迭代算法中消耗格外大，因为迭代算法常常会多次使用同一组数据。

var input = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7))
        val result = input.map(x => x*x)
        println(result.count())//7
        println(result.collect().mkString(","))//1,4,9,16,25,36,49

为了避免多次计算同一个RDD，可以让Spark 对数据进行持久化。当我们让Spark 持久化存储一个RDD 时，计算出RDD 的节点会分别保存它们所求出的分区数据。如果一个有持久化数据的节点发生故障，Spark 会在需要用到缓存的数据时重算丢失的数据分区。如果希望节点故障的情况不会拖累我们的执行速度，也可以把数据备份到多个节点上。

出于不同的目的，我们可以为RDD 选择不同的持久化级别。在Scala，默认情况下persist() 会把数据以序列化的形式缓存在JVM 的堆空间中。在Python 中，我们会始终序列化要持久化存储的数据，所以持久化级别默认值就是以序列化后的对象存储在JVM 堆空间中。当我们把数据写到磁盘或者堆外存储上时，也总是使用序列化后的数据。

org.apache.spark.storage.StorageLevel和pyspark.StorageLevel中的持久化级别；

如有必要，可以通过在存储级别的末尾加上“\_2”来把持久化数据存为两份

<table> 
 <tbody> 
  <tr> 
   <td><strong>级别</strong></td> 
   <td><strong>使用空间</strong></td> 
   <td><strong>CPU时间</strong></td> 
   <td><strong>是否在内存中</strong></td> 
   <td><strong>是否在磁盘上</strong></td> 
   <td><strong>备注</strong></td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY</td> 
   <td>高</td> 
   <td>低</td> 
   <td>是</td> 
   <td>否</td> 
   <td>　</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_2</td> 
   <td>高</td> 
   <td>低</td> 
   <td>是</td> 
   <td>否</td> 
   <td>数据存2份</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_SER</td> 
   <td>低</td> 
   <td>高</td> 
   <td>是</td> 
   <td>否</td> 
   <td>数据序列化</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_SER_2</td> 
   <td>低</td> 
   <td>高</td> 
   <td>是</td> 
   <td>否</td> 
   <td>数据序列化，数据存2份</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK</td> 
   <td>高</td> 
   <td>中等</td> 
   <td>部分</td> 
   <td>部分</td> 
   <td>如果数据在内存中放不下，则溢写到磁盘</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK_2</td> 
   <td>高</td> 
   <td>中等</td> 
   <td>部分</td> 
   <td>部分</td> 
   <td>数据存2份</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK_SER</td> 
   <td>低</td> 
   <td>高</td> 
   <td>部分</td> 
   <td>部分</td> 
   <td>　</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK_SER_2</td> 
   <td>低</td> 
   <td>高</td> 
   <td>部分</td> 
   <td>部分</td> 
   <td>数据存2份</td> 
  </tr> 
  <tr> 
   <td>DISK_ONLY</td> 
   <td>低</td> 
   <td>高</td> 
   <td>否</td> 
   <td>是</td> 
   <td>　</td> 
  </tr> 
  <tr> 
   <td>DISK_ONLY_2</td> 
   <td>低</td> 
   <td>高</td> 
   <td>否</td> 
   <td>是</td> 
   <td>数据存2份</td> 
  </tr> 
  <tr> 
   <td>NONE</td> 
   <td>　</td> 
   <td>　</td> 
   <td>　</td> 
   <td>　</td> 
   <td>　</td> 
  </tr> 
  <tr> 
   <td>OFF_HEAP&nbsp;</td> 
   <td>　</td> 
   <td>　</td> 
   <td>　</td> 
   <td>　</td> 
   <td>　</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td>MEMORY_ONLY</td> 
   <td>默认选项，RDD的（分区）数据直接以Java对象的形式存储于JVM的内存中，如果内存空间不足，某些分区的数据将不会被缓存，需要在使用的时候根据世代信息重新计算。</td> 
  </tr> 
  <tr> 
   <td>MYMORY_AND_DISK</td> 
   <td>RDD的数据直接以Java对象的形式存储于JVM的内存中，如果内存空间不中，某些分区的数据会被存储至磁盘，使用的时候从磁盘读取。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_SER</td> 
   <td>RDD的数据（Java对象）序列化之后存储于JVM的内存中（一个分区的数据为内存中的一个字节数组），相比于MEMORY_ONLY能够有效节约内存空间（特别是使用一个快速序列化工具的情况下），但读取数据时需要更多的CPU开销；如果内存空间不足，处理方式与MEMORY_ONLY相同。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK_SER</td> 
   <td>相比于MEMORY_ONLY_SER，在内存空间不足的情况下，将序列化之后的数据存储于磁盘。</td> 
  </tr> 
  <tr> 
   <td>DISK_ONLY</td> 
   <td>仅仅使用磁盘存储RDD的数据（未经序列化）。</td> 
  </tr> 
  <tr> 
   <td> <p>MEMORY_ONLY_2,</p> <p>MEMORY_AND_DISK_2, etc.</p> </td> 
   <td>以MEMORY_ONLY_2为例，MEMORY_ONLY_2相比于MEMORY_ONLY存储数据的方式是相同的，不同的是会将数据备份到集群中两个不同的节点，其余情况类似。</td> 
  </tr> 
  <tr> 
   <td>OFF_HEAP(experimental)</td> 
   <td>RDD的数据序例化之后存储至Tachyon。相比于MEMORY_ONLY_SER，OFF_HEAP能够减少垃圾回收开销、使得Spark Executor更“小”更“轻”的同时可以共享内存；而且数据存储于Tachyon中，Spark集群节点故障并不会造成数据丢失，因此这种方式在“大”内存或多并发应用的场景下是很有吸引力的。需要注意的是，Tachyon并不直接包含于Spark的体系之内，需要选择合适的版本进行部署；它的数据是以“块”为单位进行管理的，这些块可以根据一定的算法被丢弃，且不会被重建。</td> 
  </tr> 
 </tbody> 
</table>

import org.apache.spark.storage.StorageLevel
        var input = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7))
        val result = input.map(x => x * x)
        result.persist(StorageLevel.DISK_ONLY)
        println(result.count()) // 7
        println(result.collect().mkString(",")) // 1,4,9,16,25,36,49

注意，我们在第一次对这个RDD 调用行动操作前就调用了persist() 方法。persist() 调用本身不会触发强制求值。

如果要缓存的数据太多，内存中放不下，Spark 会自动利用最近最少使用（LRU）的缓存策略把最老的分区从内存中移除。对于仅把数据存放在内存中的缓存级别，下一次要用到已经被移除的分区时，这些分区就需要重新计算。但是对于使用内存与磁盘的缓存级别的分区来说，被移除的分区都会写入磁盘。不论哪一种情况，都不必担心你的作业因为缓存了太多数据而被打断。不过，缓存不必要的数据会导致有用的数据被移出内存，带来更多重算的时间开销。

最后，RDD 还有一个方法叫作unpersist()，调用该方法可以手动把持久化的RDD 从缓存中移除。