spark源碼-Checkpoint相關(guān)分四個步部分

Checkpoint相關(guān)源碼分四個步部分1，Checkpoint的基本使用:coreStreaming2，Checkpoint可以還原。輔助Spark應(yīng)用從故障中恢復(fù)。SparkStreaming宕機(jī)恢復(fù)，關(guān)鍵點進(jìn)行Checkpoint，便于故障恢復(fù)。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數(shù)據(jù)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