spark checkpoint詳解

checkpoint在spark中主要有兩塊應用：一塊是在spark core中對RDD作checkpoint，能夠切斷作checkpoint RDD的依賴關係，將RDD數據保存到可靠存儲（如HDFS）以便數據恢復；另一塊是應用在spark streaming中，使用checkpoint用來保存DStreamGraph以及相關配置信息，以便在Driver崩潰重啓的時候可以接着以前進度繼續進行處理（如以前waiting batch的job會在重啓後繼續處理）。

本文主要將詳細分析checkpoint在以上兩種場景的讀寫過程。

1，spark core中checkpoint分析

1.1，checkpoint的使用方法

使用checkpoint對RDD作快照大致以下：

sc.setCheckpointDir(checkpointDir.toString)
val rdd = sc.makeRDD(1 to 20, numSlices = 1)
rdd.checkpoint()

首先，設置checkpoint的目錄（通常是hdfs目錄），這個目錄用來將RDD相關的數據（包括每一個partition實際數據，以及partitioner（若是有的話））。而後在RDD上調用checkpoint的方法便可。

1.2，checkpoint寫流程

能夠看到checkpoint使用很是簡單，設置checkpoint目錄，而後調用RDD的checkpoint方法。針對checkpoint的寫入流程，主要有如下四個問題：

Q1：RDD中的數據是何時寫入的？是在rdd調用checkpoint方法時候嗎？

Q2：在作checkpoint的時候，具體寫入了哪些數據到HDFS了？

Q3：在對RDD作完checkpoint之後，對作RDD的本省又作了哪些收尾工做？

Q4：實際過程當中，使用RDD作checkpoint的時候須要注意什麼問題？

弄清楚了以上四個問題，我想對checkpoint的寫過程也就基本清楚了。接下來將一一回答上面提出的問題。

A1：首先看一下RDD中checkpoint方法，能夠看到在該方法中是隻是新建了一個ReliableRDDCheckpintData的對象，並無作實際的寫入工做。實際觸發寫入的時機是在runJob生成改RDD後，調用RDD的doCheckpoint方法來作的。

A2：在經歷調用RDD.doCheckpoint → RDDCheckpintData.checkpoint → ReliableRDDCheckpintData.doCheckpoint → ReliableRDDCheckpintData.writeRDDToCheckpointDirectory後，在writeRDDToCheckpointDirectory方法中能夠看到：將做爲一個單獨的任務（RunJob）將RDD中每一個parition的數據依次寫入到checkpoint目錄（writePartitionToCheckpointFile），此外若是該RDD中的partitioner若是不爲空，則也會將該對象序列化後存儲到checkpoint目錄。因此，在作checkpoint的時候，寫入的hdfs中的數據主要包括：RDD中每一個parition的實際數據，以及可能的partitioner對象（writePartitionerToCheckpointDir）。

A3：在寫完checkpoint數據到hdfs之後，將會調用rdd的markCheckpoined方法，主要斬斷該rdd的對上游的依賴，以及將paritions置空等操做。

A4：經過A1，A2能夠知道，在RDD計算完畢後，會再次經過RunJob將每一個partition數據保存到HDFS。這樣RDD將會計算兩次，因此爲了不此類狀況，最好將RDD進行cache。即1.1中rdd的推薦使用方法以下：

sc.setCheckpointDir(checkpointDir.toString)
val rdd = sc.makeRDD(1 to 20, numSlices = 1)
rdd.cache()

rdd.checkpoint()

1.3，checkpoint 讀流程

在作完checkpoint後，獲取原來RDD的依賴以及partitions數據都將從CheckpointRDD中獲取。也就是說獲取原來rdd中每一個partition數據以及partitioner等對象，都將轉移到CheckPointRDD中。

在CheckPointRDD的一個具體實現ReliableRDDCheckpintRDD中的compute方法中能夠看到，將會從hdfs的checkpoint目錄中恢復以前寫入的partition數據。而partitioner對象（若是有）也會從以前寫入hdfs的paritioner對象恢復。

總的來講，checkpoint讀取過程是比較簡單的。

2，spark streaming中checkpoint分析

2.1，streaming中checkpoint的使用方法

在streaming中使用checkpoint主要包含如下兩點：設置checkpoint目錄，初始化StreamingContext時調用getOrCreate方法，即當checkpoint目錄沒有數據時，則新建streamingContext實例，而且設置checkpoint目錄，不然從checkpoint目錄中讀取相關配置和數據建立streamingcontext。

// Function to create and setup a new StreamingContext def functionToCreateContext(): StreamingContext = { val ssc = new StreamingContext(...) // new context val lines = ssc.socketTextStream(...) // create DStreams ... ssc.checkpoint(checkpointDirectory) // set checkpoint directory ssc }

// Get StreamingContext from checkpoint data or create a new one val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)

2.2，streaming中checkpoint寫流程

一樣，針對streaming中checkpoint的寫流程，主要有如下三個問題，並對此作相關解釋。

Q1：streaming中checkpoint是在什麼時候作的？

A1：在spark streaming中，jobGenerator會按期生成任務（jobGenerator.generateJobs)。在任務生成後將會調用doCheckpoint方法對系統作checkpoint。此外，在當前批次任務結束，清理metadata（jobGenerator.clearMetadata）時，也會調用doCheckpoint方法。

Q2：在streaming checkpoint過程當中，具體都寫入了哪些數據到checkpoint目錄？

A2: 作checkpoint的主要邏輯基本都在JobGenerator.doCheckpoint方法中。

在該方法中，首先更新當前時間段須要作checkpoint RDD的相關信息，如在DirectKafkaInputDStream中，將已經生成的RDD信息的時間，topic，partition，offset等相關信息進行更新。

其次，經過checkpointWriter將Checkpoint對象寫入到checkpoint目錄中（CheckPoint.write → CheckpointWriteHandle）。至此，咱們清楚了，寫入到checkpoint目錄的數據其實就是Checkpoint對象。

Checkpoint主要包含的信息以下：

val master = ssc.sc.master
val framework = ssc.sc.appName
val jars = ssc.sc.jars
val graph = ssc.graph
val checkpointDir = ssc.checkpointDir
val checkpointDuration = ssc.checkpointDuration
val pendingTimes = ssc.scheduler.getPendingTimes().toArray
val sparkConfPairs = ssc.conf.getAll

具體包括相關配置信息，checkpoint目錄，DStreamGraph等。對於DStreamGraph，主要包含InputDstream以及outputStream等相關信息，從而咱們能夠看出定義應用相關的計算函數也被序列化保存到checkpoint目錄中了。

Q3:  streaming checkpoint都有哪些坑？

A3：

從A2中能夠看到，應用定義的計算函數也被序列化到checkpoint目錄，當應用代碼發生改變時，此時就無法從checkpoint恢復。我的感受這是checkpoint在生產環境使用中碰到的最大障礙。

另外，當從checkpoint目錄恢復streamingContext時，配置信息啥的也都是從checkpoint讀取的（只有不多的一部分配置是reload的，具體見讀流程），當重啓任務時，新改變的配置就可能不生效，致使很奇怪的問題。

此外，broadcast變量在checkpoint中使用也受到限制（SPARK-5206）。

2.3，streaming中checkpoint讀流程

在spark streaming任務從checkpoint恢復streamingContext時，將會觸發對以前保存的checkpoint對象的讀取動做。在StreamingContext的getOrCreate方法中，經過checkpoint.read方法從checkpoint目錄中恢復以前保存的Checkpoint對象。一旦該對象存在，將使用Checkpoint建立streamingContext。於此同時，在StreamingContext中DStreamGraph的恢復藉助以前保存的對象，而且調用restoreCheckpointData恢復以前生成而未計算的RDD，從而接着以前的進度進行數據處理。

另外須要注意的時，如下配置信息在使用checkpoint建立streamingContext時，這些配置信息是從新加載的。

val propertiesToReload = List(
  "spark.yarn.app.id",
  "spark.yarn.app.attemptId",
  "spark.driver.host",
  "spark.driver.bindAddress",
  "spark.driver.port",
  "spark.master",
  "spark.yarn.jars",
  "spark.yarn.keytab",
  "spark.yarn.principal",
  "spark.yarn.credentials.file",
  "spark.yarn.credentials.renewalTime",
  "spark.yarn.credentials.updateTime",
  "spark.ui.filters",
  "spark.mesos.driver.frameworkId")

3，小結

本文主要分析了checkpoint在spark core和streaming讀寫的基本過程，而且指出了在checkpoint使用中碰到一些坑。對於spark streaming，我的認爲checkpoint在生產環境不適宜使用。