什么是Checkpointing
Checkpointing可以将RDD从其依赖关系中抽出来,保存到可靠的存储系统(例如HDFS,S3等), 即它可以将数据和元数据保存到检查指向目录中。 因此,在程序发生崩溃的时候,Spark可以恢复此数据,并从停止的任何地方开始。
Checkpointing分为两类:
- reliable,容错性优先。这种类型的检查点可确保数据永久存储,如在HDFS或其他分布式文件系统上。 这也意味着数据通常会在网络中复制,这会降低检查点的运行速度
- local,性能优先。 RDD持久保存到执行程序中的本地存储。 因此,数据写得更快,但本地存储也不是完全可靠的,一旦数据丢失,工作将无法恢复。==一般用于需要定期截取且拥有较长的lineage关系的RDD==,例如,GraphX。
开发人员可以是来 ==RDD.checkpoint()== 方法来设置检查点。
在使用检查点之前,必须使用 SparkContext.setCheckpointDir(directory: String) 方法设置检查点目录。
<font color=darkb><b>Checkpointing waits until the end of a job, and launches another job to finish checkpoint. An RDD which needs to be checkpointed will be computed twice; thus it is suggested to do a rdd.cache() before rdd.checkpoint().
强烈建议在启用RDD检查点之前,先将其persist到内存中,否则,在它存到文件之前,需要重新计算一次。
为什么使用Checkpointing
RDD的检查点机制就好比Hadoop将中间计算值存储到磁盘,即使计算中出现了故障,我们也可以轻松地从中恢复。通过对 RDD 启动检查点机制可以实现容错和高可用。
在Spark Streaming程序中,如果某些数据已经在队列中等待处理,由于某些原因我们的应用程序崩溃,当我们再次启动时,则无需再次读取这些数据,并且数据不会丢失。
如果我们的应用程序正在使用任何有状态操作,那么检查点是必需的,否则一旦应用程序崩溃,所有状态都将丢失。
哪些RDD需要使用Checkpointing
- 计算需要很长时间
- 计算链太长了
- 依赖于于太多的父RDD
Checkpointing和Cache的区别
Cache和Checkpointing之间存在显着差异。Cache可以物理化RDD并将其保留在内存(和/或磁盘)中。但是它会==记住RDD的依赖关系(lineage)==,因此就算存在节点故障导致部分缓存的RDD丢失,它们可以被重新生成。
Checkpointing仅将RDD数据保存到HDFS文件中,而==忽略依赖关系==。
此外,rdd.persist(StorageLevel.DISK_ONLY)也与checkpoint不同。通过前者可以将RDD分区持久化到磁盘,分区由blockManager管理。一旦驱动程序完成,这意味着CoarseGrainedExecutorBackend所在的线程停止,blockManager将停止,缓存到磁盘的RDD将被删除(blockManager使用的本地文件将被删除)。但检查点会将RDD保留到HDFS或本地目录。如果没有手动删除,它们将始终位于磁盘上,因此下一个驱动程序可以使用它们。
参考资料
cache and checkpoint
Checkpointing in Spark
Apache Spark: Caching and Checkpointing Under the Hood
Spark RDD Checkpointing