《Spark Streaming实时流式大数据处理实战》 ——3.6 RDD持久化（Cachinng/Persistence

3.6  RDD持久化（Cachinng/Persistence）

　　前面几节介绍了RDD的各种细节，本节将介绍Spark赋予RDD的另一个特性，即持久化（Persisting/Cache）。这个概念其实很好理解，我们在前文中介绍过RDD通过转换操作（Transformation）会形成有向无环图，之后Action操作会激活真实的计算。

　　如果我们持久化一个RDD，其每个节点的各个分支（partition）会将计算结果保存在内存中，并可以将其用在其他Action操作中，也就是说我们只需计算一次RDD，在将来有Action操作需要再次用到该RDD的时候，速度会更加快（通常超过10倍）。将RDD持久化在迭代算法和快速交互场景中可以起到关键作用。

　　我们可以通过rdd.persist()或者rdd.cache()方法将一个RDD标记为持久化，一旦一个RDD通过Action操作激活计算后，其就会保存在节点的内存中。并且Spark的缓存机制是有容错性的，即如果一个RDD的某个分支（Partition）丢失，它会自动根据之前创建的转换操作重新计算。

　　此外，可以使用不同的存储级别（storage level）对RDD进行持久化，在节点之间复制。例如，可以将一个数据集持久化在硬盘或者内存，就像Java当中对对象的序列化（serialized）一样。在调用persist方法时，我们可以传递存储级别参数来进行设置，而cache方法会使用默认的存储级别StorageLevel.MEMORY_ONLY（在内存中保存和反序列化对象）。所有的存储级别如表3.2所示。

表3.2  RDD持久化的不同存储级别

以序列化的格式存储RDD到离线堆内存中（OFF_HEAP Memory），要求OFF_HEAP Memory是可用的。相对于MEMORY_ONLY_SER，OFF_HEAP减少了垃圾回收的花费，允许更小的执行者共享内存池。这使其在拥有大量内存的环境下或者多并发应用程序的环境中具有更强的吸引力

?注意：当通过Python使用Spark时，所有需要存储的对象都会由Pickle库来完成序列化，所以对于Python版本序列化级别（serialized level）是无关紧要的。Python中可选择的存储级别（storage level）包括：MEMORY_ONLY、MEMORY_ONLY_2、MEMORY_AND_DISK、MEMORY_AND_DISK_2、DISK_ONLY和 DISK_ONLY_2。

　　另外，即使用户没有调用persist函数，Spark也会自动将Shuffle操作的一些中间数据进行持久化（如reduceByKey）。这主要是为了防止在Shuffle操作期间，当一个节点失败时导致全局的重新计算。对于需要反复使用的RDD结果，最好利用persist操作将其持久化，从而避免重复计算。

　　关于如何选择存储级别，需要权衡内存消耗和CPU效率，主要从以下几点来考虑：

* 如果在默认的存储级别（MEMORY_ONLY）满足要求的情况下，就不要切换，因为这是CPU的最高效形式，可以使得RDD操作尽可能地快速执行。

* 如果不能满足要求，那么尝试MEMORY_ONLY_SER，并且选取一个高效的序列化库使得对象能够在空间合理、高效访问的前提下被序列化（只能在Java和Scala中使用）。

* 除非计算RDD的花费较大或者它们需要过滤大量的数据，不要将RDD存储到磁盘上（DISK），否则，重复计算一个分区就和从磁盘上读取数据一样慢。

* 如果希望更快的错误恢复（如将Spark用来服务一个Web应用的请求），可以利用重复（replicated）存储级别。所有的存储级别都可以通过重复计算丢失的数据来支持完整的容错，但是重复的数据能够使我们在RDD上继续运行任务，而不需要重复计算丢失的数据。

　　Spark会自动监视每个节点的缓存使用情况，并且根据最近最久未使用（LRU least-recently-used）原则来删除旧的数据分支。如果希望手动将RDD从缓存中移除，可以使用RDD.unpersist方法。