|
? 首先，这三者都是做RDD持久化的，cache()和persist()是将数据默认缓存在内存中，checkpoint()是将数据做物理存储的（本地磁盘或Hdfs上），当然rdd.persist(StorageLevel.DISK_ONLY)也可以存储在磁盘 。
? 其次，缓存机制里的cache和persist都是用于将一个RDD进行缓存，区别就是：cache()是persisit()的一种简化方式，cache()的底层就是调用的persist()的无参版本，同时就是调用persist(MEMORY_ONLY)将数据持久化到内存中。如果需要从内存中清除缓存，那么可以使用unpersist()方法。cache () = persist()=persist(StorageLevel.Memory_Only)
? 另外，cache 跟 persist不会截断血缘关系，checkPoint会截断血缘关系。
cache()与persist()的区别
/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
def cache(): this.type = persist()
/**
* 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. Local checkpointing is an exception.
*/
def persist(newLevel: StorageLevel): this.type = {
if (isLocallyCheckpointed) {
// This means the user previously called localCheckpoint(), which should have already
// marked this RDD for persisting. Here we should override the old storage level with
// one that is explicitly requested by the user (after adapting it to use disk).
persist(LocalRDDCheckpointData.transformStorageLevel(newLevel), allowOverride = true)
} else {
persist(newLevel, allowOverride = false)
}
}
/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)
? 从以上源码可以看出，cache 底层调用的是 persist 方法，存储等级为: memory only，persist 的默认存储级别也是 memory only，persist 与 cache 的主要区别是 persist 可以自定义存储级别。如rdd.persist(StorageLevel.DISK_ONLY) ，就可以将数据存储在磁盘上。哪些 RDD 需要cache？会被重复使用的(但是)不能太大的RDD需要cache，cache 只使用 memory。
cache()与checkpoint()的区别
? cache 和 checkpoint 之间有一个重大的区别，cache 将 RDD 以及 RDD 的血统(记录了这个RDD如何产生)缓存到内存中，当缓存的 RDD 失效的时候(如内存损坏)，它们可以通过血统重新计算来进行恢复。但是 checkpoint 将 RDD 缓存到了 HDFS 中，同时忽略了它的血统(也就是RDD之前的那些依赖)。为什么要丢掉依赖？因为可以利用 HDFS 多副本特性保证容错！
/**
* 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.
*/
def checkpoint(): Unit = RDDCheckpointData.synchronized {
// NOTE: we use a global lock here due to complexities downstream with ensuring
// children RDD partitions point to the correct parent partitions. In the future
// we should revisit this consideration.
if (context.checkpointDir.isEmpty) {
throw new SparkException("Checkpoint directory has not been set in the SparkContext")
} else if (checkpointData.isEmpty) {
checkpointData = Some(new ReliableRDDCheckpointData(this))
}
}
? 从上面的源码中可以看出，在进行checkpoint之前要设置 sc.setCheckpointDir("...")。除此之外，在进行 checkpoint 前，要先对 RDD 进行 cache。Why??? checkpoint 会等到 job 结束后另外启动专门的 job 去完成 checkpoint，也就是说需要 checkpoint 的 RDD 会被计算两次。
persist()与checkpoint()的区别
persist(StorageLevel.DISK_Only) != checkPoint()
? rdd.persist(StorageLevel.DISK_ONLY) 与 checkpoint 也有区别。前者虽然可以将 RDD 的 partition 持久化到磁盘，但该 partition 由 blockManager 管理。一旦 driver program 执行结束，也就是 executor 所在进程 CoarseGrainedExecutorBackend stop，blockManager 也会 stop，被 cache 到磁盘上的 RDD 也会被清空（整个 blockManager 使用的 local 文件夹被删除）。而 checkpoint 将 RDD 持久化到 HDFS 或本地文件夹，如果不被手动 remove 掉，是一直存在的，也就是说可以被下一个 driver program 使用，而 cached RDD 不能被其他 dirver program 使用。
应用场景：
cache: 对于会被重复使用，但是数据量不是太大的RDD，可以将其cache（）到内存当中。cache 是每计算出一个要 cache 的 partition 就直接将其 cache 到内存中。缓存完之后，可以在任务监控界面storage里面看到缓存的数据。
checkpoint：对于computing chain 计算链过长或依赖其他 RDD 很多的 RDD，就需要进行checkpoint，将其放入到HDFS或者本地文件夹当中。需要注意的是，checkpoint 需要等到job完成了，再启动专门的job去完成checkpoint 操作，因此RDD是被计算了两次的。一般使用的时候配合rdd.cache()，这样第二次就不用重新计算RDD了，直接读取 cache 写磁盘。
注意：rdd.persist(StorageLevel.DISK_ONLY) 与 checkpoint 也有区别，persist一旦程序执行结束，所有的缓存无论在内存还是磁盘都会被删掉。而 checkpoint 将 RDD 持久化到 HDFS 或本地文件夹，如果不被手动 remove 掉，是一直存在的，也就是说可以被下一个 driver，program 使用，而 cached RDD 不能被其他 dirver program 使用。
spark的缓存级别参照【org.apache.spark.storage.StorageLevel.scala】
? ? 在persist()中可以指定一个StorageLevel,当StorageLevel为MEMORY_ONLY时就是cache. ? ??
? ? StorageLevel的列表可以在StorageLevel伴生单例对象中查看 : ?
class StorageLevel private(
private var _useDisk: Boolean,
private var _useMemory: Boolean,
private var _useOffHeap: Boolean,
private var _deserialized: Boolean,
private var _replication: Int = 1) //hdfs的副本是存3份，RDD默认的副本数是1
extends Externalizable
new StorageLevel(_useDisk,_useMemory, _useOffHeap,_deserialized,_replication: Int = 1)
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(true, true, true, false, 1)
- Memory_Only:只存于内存
- Disk_Only:只存于磁盘
- Memory_and_disk:内存不够，存于磁盘
- Memory_only_ser:将数据以序列化的方式存于内存? ?
? ? 默认缓存级别：def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)，
? ? 默认情况下persist() 会把数据以反序列化的形式缓存在JVM的堆（heap）空间中。?
? ? 取消缓存，执行RDD.unpersist()? ? ? ? ? ? ? ? ? ? ?
Examples:
1.缓存cache()的使用
object Test01 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setAppName("Demo").setMaster("local"))
val rdd1 = sc.textFile("file:///D:\\IdeaProgram\\Bigdata\\file\\Spark\\wc\\input")
//在这里设置一个cache缓存,只缓存结果,建议缓存完之后执行action,再去执行转换算子+执行算子
val rdd2 = rdd1.flatMap(_.split("\\s+")).map((_, 1)).reduceByKey(_ + _).cache()
//这里如果不设置缓存,每次执行完一个action以后再去执行下一个action的时候都要从头开始进行计算
//如果设置了缓存,每次执行完一个action以后再去执行下一个action的时候直接从缓存为止读取数据
rdd2.collect()
rdd2.foreach(println(_))
rdd2.take(2)
}
}
2.persist()的使用
import org.apache.spark.storage._
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("DISK_ONLY")
val sc = new SparkContext(conf);
sc.setLogLevel("ERROR")
val a = sc.parallelize(1 to 9, 3).persist(StorageLevel.DISK_ONLY)
println(a.first())
}
一旦 driver program 执行结束，也就是 executor 所在进程 CoarseGrainedExecutorBackend stop，blockManager也会stop；被缓存到磁盘上的RDD也会被清空（整个blockManager使用的local文件夹被删除）
3.checkpoint()的使用
checkpoint检查点机制
检查点(本质就是通过将RDD写入Disk做检查点)是为了通过lineage(血统)做容错的辅助,lineage过长会造成容错成本过高,这样就不如在中间阶段做检查点容错,如果之后有节点问题而丢失分区,从做查点的RDD开始重做lineage,就会减少开销检查点通过将数据写入到HDFS文件系统实现了RDD的检查点功能.
检查点,类似于快照,chekpoint的作?就是将DAG中?较重要的数据做?个检查点,将结果存储到?个?可?的地?
//检查点的使用
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("SparkDemo").setMaster("local")
val sc = new SparkContext(conf)
//设置检查点的路径
sc.setCheckpointDir("hdfs://hadoop01:8020/ck")
val rdd = sc.textFile("hdfs://hadoop01:8020/word.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
//检查点的触发?定要使?个action算?
rdd.checkpoint()
rdd.saveAsTextFile("hdfs://hadoop01:8020/out10")
println(rdd.getCheckpointFile) //查看存储的位置
/**
查看是否可以设置检查点 rdd.isCheckpointed 这个?法在shell中可以使? 但是代码中不好?
*/
}
4.实用心得：
缓存和检查点的区别
缓存把RDD计算出来的然后放在内存,但是RDD的依赖链不能丢掉,当某个exexutor宕机时,上面cache的RDD就会丢掉,需要通过依赖链放入重新计算,不同的是,checkpoint是把RDD保存在HDFS上,是多副本可靠存储,所以依赖链可以丢掉,就是斩断了依赖链,是通过复制实现的高容错
cache和persist的比较
1.cache底层调用的是persist
2.cache默认持久化等级是内存且不能修改,persist可以修改持久化的等级
什么时候使用cache或checkpoint
1.某步骤计算特别耗时
2.计算链条特别长
3.发生shuffle之后
一般情况建议使用cache或是persist模式,因为不需要创建存储位置,默认存储到内存中计算速度快,而checkpoint需要手动创建存储位置和手动删除数据,若数据量非常庞大建议使用checkpoint
Task包括ResultTask(最后一个阶段的任务) + ShuffleMapTask(非最后一个阶段)
|