Spark RDD概念学习系列之RDD的checkpoint(九)-阿里云开发者社区

开发者社区> 大数据> 正文
登录阅读全文

Spark RDD概念学习系列之RDD的checkpoint(九)

简介:

RDD的检查点

  首先,要清楚。为什么spark要引入检查点机制?引入RDD的检查点?

   如果缓存丢失了,则需要重新计算。如果计算特别复杂或者计算耗时特别多,那么缓存丢失对于整个Job的影响是不容忽视的。为了避免缓存丢失重新计算带来的开销,Spark又引入检查点机制。

 

 

      RDD的缓存能够在第一次计算完成后,将计算结果保存到内存、本地文件系统或者Tachyon(分布式内存文件系统)中。通过缓存,Spark避免了RDD上的重复计算,能够极大地提升计算速度。但是,如果缓存丢失了,则需要重新计算。如果计算特别复杂或者计算耗时特别多,那么缓存丢失对于整个Job的影响是不容忽视的。为了避免缓存丢失重新计算带来的开销,Spark又引入检查点(checkpoint)机制。

 

 

 

 

RDD的缓存和RDD的checkpoint的区别

      RDD的缓存是在计算结束后,直接将计算结果通过用户定义的存储级别(存储级别定义了缓存存储的介质,现在支持内存、本地文件系统和Tachyon)写入不同的介质。

      而RDD的检查点不同,它是在计算完成后,重新建立一个Job来计算。

      为了避免重复计算,推荐先将RDD缓存,这样就能保证检查点的操作可以快速完成。

    

 

 

 

 

 

RDD的checkpoint的处理

  在缓存没有命中的情况下,首先会判断是否保存了RDD的checkpoint,如果有,则读取checkpoint。为了理解checkpoint的RDD是如何读取计算结果的,需要先看一下checkpoint的数据是如何写入的。
首先在Job结束后,会判断是否需要checkpoint。如果需要,就调用org.apache.spark.rdd.RDDCheckpointData#doCheckpoint。doCheckpoint首先为数据创建一个目录;然后启动一个新的Job来计算,并且将计算结果写入新创建的目录;接着创建一个org.apache.spark.rdd.CheckpointRDD;最后,原始RDD的所有依赖被清除,这就意味着RDD的转换的计算链(compute chain)等信息都被清除。这个处理逻辑中,数据写入的实现在org.apache.spark.rdd.CheckpointRDD$#writeToFile。

 

简要的核心逻辑如下:

复制代码
//创建一个保存checkpoint数据的目录
val path = new Path(rdd.context.checkpointDir.get, "rdd-" + rdd.id)
val fs = path.getFileSystem(rdd.context.hadoopConfiguration)

if (!fs.mkdirs(path)) {
throw new SparkException("Failed to create checkpoint path " + path)
}
//创建广播变量 val broadcastedConf = rdd.context.broadcast( new SerializableWritable(rdd.context.hadoopConfiguration))
//开始一个新的Job进行计算,计算结果存入路径path中 rdd.context.runJob(rdd, CheckpointRDD.writeToFile[T](path.toString, broadcastedConf) _)
//根据结果的路径path来创建CheckpointRDD val newRDD = new CheckpointRDD[T](rdd.context, path.toString)
//保存结果,清除原始RDD的依赖、Partition信息等 RDDCheckpointData.synchronized { cpFile = Some(path.toString) cpRDD = Some(newRDD) // RDDCheckpointData对应的CheckpointRDD rdd.markCheckpointed(newRDD) //清除原始RDD的依赖,Partition cpState = Checkpointed //标记checkpoint的状态为完成 }
复制代码

  至此,RDD的checkpoint完成,其中checkpoint的数据可以通过checkpointRDD的readFromFile读取。但是,上述逻辑在清除了RDD的依赖后,并没有和check-pointRDD建立联系。

 

 

 

 

 

 

  那么Spark是如何确定一个RDD是否被checkpoint了,而且正确读取checkpoint的数据呢?
答案就在org.apache.spark.rdd.RDD#dependencies的实现,它会首先判断当前的RDD是否已经Checkpoint过,如果有,那么RDD的依赖就变成了对应的Ch

复制代码
eckpointRDD:

privatedefcheckpointRDD: Option[RDD[T]]=checkpointData.flatMap(_.checkpointRDD)
final def dependencies: Seq[Dependency[_]] = {
checkpointRDD.map(r => List(new OneToOneDependency(r))).getOrElse {
if (dependencies_ == null) { //没有checkpoint
dependencies_ = getDependencies
}

 

dependencies_
}
}
复制代码

  理解了Checkpoint的实现过程。

 

 

 

  接下来看一下computeOrReadCheckpoint的实现。前面提到了,它一共在两个地方被调用,org.apache.spark.rdd.RDD#iterator和org.apache.spark.CacheManager#getOrCompute。它实现的逻辑比较简单,首先检查当前RDD是否被Checkpoint过,如果有,读取Checkpoint的数据;否则开始计算。

 

实现如下:

复制代码
private[spark] def computeOrReadCheckpoint(split: Partition, context: TaskContext)
: Iterator[T] =

 

{
if (isCheckpointed) firstParent[T].iterator(split, context) else compute(split, context)
}
firstParent[T].iterator(split,context)会调用对应CheckpointRDD的iterator,最终调用到它的compute:
override def compute(split: Partition, context: TaskContext): Iterator[T] = {
val file=new Path(checkpointPath, CheckpointRDD.splitIdToFile(split.index))
CheckpointRDD.readFromFile(file, broadcastedConf, context) //

 
读取Checkpoint的数据
}
复制代码

 

 


本文转自大数据躺过的坑博客园博客,原文链接:http://www.cnblogs.com/zlslch/p/5723758.html,如需转载请自行联系原作者

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享: