1. Background

当我们需要多次使用同一个 RDD 时，如果简单的调用 Action 操作，Spark 每次都会重算 RDD 以及它所有的依赖，此时需要用到持久化技术。

2. Basic

2.1 持久化级别

Storage Level	Meaning
MEMORY_ONLY	未序列化只保存在内存，如果内存装不下，那么一部分分区数据将在他们被需要的时候重新计算。该级别是默认持久化级别
MEMORY_AND_DISK	未序列化先尝试保存在内存，如果内存装不下，那么一部分分区数据保存在磁盘
MEMORY_ONLY_SER	MEMORY_ONLY 的序列化版
MEMORY_AND_DISK_SER	MEMORY_AND_DISK 的序列化版
DISK_ONLY	只保存在磁盘
MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc.	带副本的版本

2.2 cache() & persist()

def cache() = persist()

def persist() = persist(StorageLevel.MEMORY_ONLY)

根据二者的函数实现可以看出，cache 底层就是调用的 persist 的无参重载函数，而无参重载函数的持久化级别是MEMORY_ONLY。日常使用中直接调用 cache 即可

2.3 cache() 的使用

scala> val rdd = sc.makeRDD(Array("rdd"))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[47] at makeRDD at <console>:24

scala> val noCache=rdd.map(_+System.currentTimeMillis())
noCache: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[48] at map at <console>:26

scala> noCache.collect
res33: Array[String] = Array(rdd1600019753783)

scala> noCache.collect
res34: Array[String] = Array(rdd1600019756380)

scala> noCache.collect
res35: Array[String] = Array(rdd1600019759584)

scala> val useCache=noCache.cache
useCache: noCache.type = MapPartitionsRDD[48] at map at <console>:26

scala> useCache.collect
res36: Array[String] = Array(rdd1600019789962)

scala> useCache.collect
res37: Array[String] = Array(rdd1600019789962)

scala> useCache.collect
res38: Array[String] = Array(rdd1600019789962)

从上面的代码中可以看到，在未使用 cache 前，每次的时间戳都是重新计算的。对 RDD 使用了 cache 后，则时间戳被保存了下来

3. Deep

3.1 任何时候都应该使用 cache 吗？

既然 cache 的默认持久化级别是 MEMORY_ONLY，内存装不下时，RDD 的部分分区不进行缓存。那是否意味着任何时候都应该对 RDD 进行 cache 操作呢？答案显然是不应该的。因为内存资源是有限的，归根结底，cache 操作是在重新计算 RDD 的计算代价与缓存 RDD 的内存资源之间进行权衡。如果内存资源充足，但是 cpu 计算资源紧张，那么可以多进行 cache 操作。反之则要谨慎 cache

3.2 DISK_ONLY 持久化级别有什么意义？

除了绝大多数时候对 RDD 使用的 cache 操作外，我们可以通过 persist 自己手动指定持久化级别。持久化到内存中是很容易理解的，和缓存概念类似。但是 DISK_ONLY 很反直觉的将 RDD 持久化到磁盘。这样做是否有意义呢？如果理解了前面所说的 cache 操作是在重新计算 RDD 的计算代价与缓存 RDD 的内存资源之间进行权衡的话，这里也就很好理解了。虽然磁盘资源可以认为是永远充足的，但是从磁盘读取是有 I/O 代价的，所以 DISK_ONLY 持久化级别就是在重新计算 RDD 的计算代价与从磁盘读取 RDD 的 I/O 代价之间进行权衡。换句话说如果 cpu 和内存资源不充足，并且计算该 RDD 耗时很久，可以考虑将持久化级别设置为 DISK_ONLY。

3.3 cache & checkpoint

3.3.1 checkpoint 注释：

Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint
directory set with SparkContext#setCheckpointDir and all references to its parent
RDDs will be removed. This function must be called before any job has been
executed on this RDD. It is strongly recommended that this RDD is persisted in
memory, otherwise saving it on a file will require recomputation.

从注释中可以总结出如下几个要点：

RDD 将被保存在事先通过 setCheckpointDir 设置的路径下
RDD 的血缘关系将被切断
必须在所有 action 前调用
最好将使用 checkpoint 的 RDD 持久化在内存中，否则 checkpoint 时将触发重复计算

3.3.2 代码演示：

scala> val rdd = sc.makeRDD(Array("rdd"))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at makeRDD at <console>:24

scala> val ckRdd = rdd.map(_+System.currentTimeMillis())
ckRdd: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[1] at map at <console>:26

scala> sc.setCheckpointDir("./ck")

scala> ckRdd.checkpoint

scala> ckRdd.toDebugString
res3: String =
(4) MapPartitionsRDD[1] at map at <console>:26 []
 |  ParallelCollectionRDD[0] at makeRDD at <console>:24 []

scala> ckRdd.collect
res4: Array[String] = Array(rdd1600026196973)

此时查看 WebUI：

ckRdd.collect 触发的 job

可以看到对 checkpoint 的 RDD 首次 collect 时，触发了两个 job。第一个 job 用来返回结果，第二个 job 重新计算并持久化到磁盘。

scala> ckRdd.collect
res5: Array[String] = Array(rdd1600026197161)

scala> ckRdd.collect
res6: Array[String] = Array(rdd1600026197161)

scala> ckRdd.collect
res6: Array[String] = Array(rdd1600026197161)

scala> ckRdd.toDebugString
res7: String =
(4) MapPartitionsRDD[1] at map at <console>:26 []
 |  ReliableCheckpointRDD[2] at collect at <console>:29 []

scala> val cacheRdd = rdd.map(_+System.currentTimeMillis())
cacheRdd: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[3] at map at <console>:26

scala> cacheRdd.cache
res8: cacheRdd.type = MapPartitionsRDD[3] at map at <console>:26

scala> cacheRdd.collect
res10: Array[String] = Array(rdd1600026882262)

scala> cacheRdd.toDebugString
res11: String =
(4) MapPartitionsRDD[3] at map at <console>:26 [Memory Deserialized 1x Replicated]
 |       CachedPartitions: 4; MemorySize: 144.0 B; ExternalBlockStoreSize: 0.0 B; DiskSize: 0.0 B
 |  ParallelCollectionRDD[0] at makeRDD at <console>:24 [Memory Deserialized 1x Replicated]

通过演示可以看到，checkpoint 后的 RDD 第二次 collect 的结果和第一次不同，时间戳要大一些，验证了第一次调用 collect 时触发了两个 job ，且使用的是触发的第二个 job持久化到磁盘的结果。通过调用 toDebugString 可以看到，checkpointRDD 切断了血缘关系只保留了一个ReliableCheckpointRDD，作为对比 cache RDD 则在中间加了一层 CachedPartitions，保留了完整的血缘关系。

接下来看看在使用 checkpoint 前调用 action 算子是什么效果：

scala> val rdd = sc.makeRDD(Array("rdd"))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[4] at makeRDD at <console>:24

scala> val ckRdd = rdd.map(_+System.currentTimeMillis())
ckRdd: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[5] at map at <console>:26

scala> sc.setCheckpointDir("./ck")

scala> ckRdd.collect
res13: Array[String] = Array(rdd1600027380204)

scala> ckRdd.checkpoint

scala> ckRdd.collect
res16: Array[String] = Array(rdd1600027397304)

scala> ckRdd.collect
res17: Array[String] = Array(rdd1600027399110)

scala> ckRdd.collect
res18: Array[String] = Array(rdd1600027399948)

可以看到，如果在调用 checkpoint 前调用了 action 算子，将会导致 checkpoint 失效。

3.3.3 cache & checkpoint 对比总结：

cache()：MEMORY_ONLY for RDD,MEMORY_AND_DISK for Dataset；checkpoint需要手动指定一个外部文件系统作为检查点，保存在磁盘
cache 在执行job过程中直接保存中间结果；checkpoint 会另启一个 job 重新计算
cache 可以在 action 后调用；checkpoint 必须在 action 前调用，否则失效
cache 不会切断血缘关系(在中间添加一层 cacheRDD )；checkpoint 会切断血缘关系
checkpoint 一般可以保存系统状态，用于故障恢复；cache 不具备这个特性

3.3.4 最佳实践

对 RDD 先调用 cache 保存在内存，再调用 checkpoint 持久化到磁盘，checkpoint 虽然会触发第二个 job 来持久化到磁盘，但是因为 cache 在了内存中，所以直接从内存中取，避免了重复计算。

Spark RDD 之持久化