spark源码-Checkpoint相关分四个步部分

Checkpoint相关源码分四个步部分1，Checkpoint的基本使用:coreStreaming2，Checkpoint可以还原。辅助Spark应用从故障中恢复。SparkStreaming宕机恢复，关键点进行Checkpoint，便于故障恢复。1,Cacheorvalsc=newvalvalcount=valfile=valvalcountcheckpointData=Some(new if(checkpointData.isDefined)}else}valnewRDD=//UpdateourstateandtruncatetheRDDcpRDD=cpState=}valnewRDD=ReliableCheckpointRDD.writeRDDToCheckpointDirectory(rdd, valvalsc=checkpointDirPath valfs=valvalbroadcastedConf=new//TODO:ThisisexpensivebecauseitcomputestheRDDagain(SPARK- writePartitionerToCheckpointDir(sc,}四，Checkpoint数据finaldefiterator(split:Partition,context:TaskContext):Iterator[T]=if(storageLevel!=StorageLevel.NONE) pute(this,split,context,}else}}private[spark]defcomputeOrReadCheckpoint(split:Partition,TaskContext):Iterator[T]{if(isCheckpointedAndMaterialized){}elsecompute(split,}}/**Returns parentRDDprotected/**Returns parentRDDprotected[spark]def Parent[U:ClassTag]:RDD[U]={}finaldefdependencies:Seq[Dependency[_]]=if(dependencies_==null)dependencies_=}}}privateprivatedefcheckpointRDD:Option[CheckpointRDD[T]] valvalcheckpointOption=checkpointPath,newSparkConf(),hadoopConf,StreamingContextprivate[streaming]sc_:cp_:batchDur_:)extendsLogging然后去valvalcheckpointOption=checkpointPath,newSparkConf(),hadoopConf,//TrytofindthecheckpointvalcheckpointFiles=if(checkpointFiles.isEmpty)return}checkpointFiles.foreach(file=>logInfo("Attemptingtoloadcheckpointfromfile"+tryvalfis=valcp=Checkpoint.deserialize(fis,logInfo("Checkpointsuccessfullyloadedfromfile"+logInfo("Checkpointwasgeneratedattime"+return}catchcasee:ExceptionreadError=logWarning("Errorreadingcheckpointfromfile"+file,}private[streaming]valisCheckpointPresent=(cp_!=private[streaming]valsc:SparkContext=if(sc_!=null)}elseif(isCheckpointPresent)}elsethrownewSparkException("CannotcreateStreamingContextwithout}}private[streaming]valgraph:DStreamGraph=if(isCheckpointPresent)}elserequire(batchDur_!=null,"BatchdurationforStreamingContextcannotbevalnewGraph=new}}private[streaming]varcheckpointDir:String=if(isCheckpointPresent)}else}}}}if(isCheckpointPresent)cp_.checkpointDurationelseprivate[streaming]v