spark中的map和flattop,persist和cache分别有什么区别？

ゞ浴缸里的玫瑰 2022-06-03 07:16 174阅读 0赞

首先，说一下map和flatmap：  
map函数会对每一条输入进行指定的操作，然后为每一条输入返回一个对象，对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。  
![这里写图片描述][SouthEast]

如图，flatmap是把map的结果扁平化处理，得到一个集合。

--------------------

在spark中，数据如果需要重复使用或者进行最终的存储，通常需要进行持久化，可以使用cache()和persist(),但使用两者还是有区别的。  
首先是cache()  
从源码可以看出，cache其实是调用了persist()

/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
    def cache(): this.type = persist()

再看一下源码中对persist()函数的定义：

/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
    def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

persist（）函数内部调用了persist(StorageLevel.MEMORY\_ONLY)  
然后，看一下persist(StorageLevel.MEMORY\_ONLY)：

/** * Set this RDD's storage level to persist its values across operations after the first time * it is computed. This can only be used to assign a new storage level if the RDD does not * have a storage level set yet.. */
    def persist(newLevel: StorageLevel): this.type = {
      // TODO: Handle changes of StorageLevel
      if (storageLevel != StorageLevel.NONE && newLevel != storageLevel) {
        throw new UnsupportedOperationException(
          "Cannot change storage level of an RDD after it was already assigned a level")
      }
      sc.persistRDD(this)
      // Register the RDD with the ContextCleaner for automatic GC-based cleanup
      sc.cleaner.foreach(_.registerRDDForCleanup(this))
      storageLevel = newLevel
      this
    }

发现persist(StorageLevel.MEMORY\_ONLY)有一个可设置StorageLevel类型参数，即缓存级别。  
那么RDD的缓存级别都有什么呢？查看StorageLevel的源码，发现：

object StorageLevel { 
      val NONE = new StorageLevel(false, false, false, false)
      val DISK_ONLY = new StorageLevel(true, false, false, false)
      val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
      val MEMORY_ONLY = new StorageLevel(false, true, false, true)
      val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
      val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
      val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
      val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
      val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
      val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
      val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
      val OFF_HEAP = new StorageLevel(false, false, true, false)
      ......
    }

竟然有12种缓存级别！！！所以说persist（）函数是可以在这12种缓存级别中进行选择的，而cache()只能MEMORY\_ONLY.  
继续深究一下，12种缓存级别，分别表示存在哪：

val MEMORY_ONLY = new StorageLevel(false, true, false, true)

比如，MEMORY\_ONLY有四个参数，查看其构造函数：

class StorageLevel private(
        private var _useDisk: Boolean,
        private var _useMemory: Boolean,
        private var _useOffHeap: Boolean,
        private var _deserialized: Boolean,
        private var _replication: Int = 1)
      extends Externalizable {
      ......
      def useDisk: Boolean = _useDisk
      def useMemory: Boolean = _useMemory
      def useOffHeap: Boolean = _useOffHeap
      def deserialized: Boolean = _deserialized
      def replication: Int = _replication
      ......
    }

可以看到StorageLevel类的主构造器包含了5个参数：

useDisk：使用硬盘（外存）  
useMemory：使用内存  
useOffHeap：使用堆外内存，这是Java虚拟机里面的概念，堆外内存意味着把内存对象分配在Java虚拟机的堆以外的内存，这些内存直接受操作系统管理（而不是虚拟机）。这样做的结果就是能保持一个较小的堆，以减少垃圾收集对应用的影响。  
deserialized：反序列化，其逆过程序列化（Serialization）是java提供的一种机制，将对象表示成一连串的字节；而反序列化就表示将字节恢复为对象的过程。序列化是对象永久化的一种机制，可以将对象及其属性保存起来，并能在反序列化后直接恢复这个对象  
replication：备份数（在多个节点上备份）

so,  
val MEMORY\_AND\_DISK\_SER\_2 = new StorageLevel(true, true, false, false, 2) 就表示使用这种缓存级别的RDD将存储在硬盘以及内存中，使用序列化（在硬盘中），并且在多个节点上备份2份（正常的RDD只有一份）

另外还注意到有一种特殊的缓存级别

val OFF_HEAP = new StorageLevel(false, false, true, false)

使用了堆外内存，StorageLevel 类的源码中有一段代码可以看出这个的特殊性，它不能和其它几个参数共存。

if (useOffHeap) { require(!useDisk, "Off-heap storage level does not support using disk") require(!useMemory, "Off-heap storage level does not support using heap memory") require(!deserialized, "Off-heap storage level does not support deserialized storage") require(replication == 1, "Off-heap storage level does not support multiple replication")

[SouthEast]: /images/20220603/e9572e96fed04ca88fe4e84531de793d.png