(版本定制)第13課：Spark Streaming源碼解讀之Driver容錯安全性

本期內容：
1. ReceiverBlockTracker容錯安全性 
2. DStream和JobGenerator容錯安全性

一：容錯安全性 
1. ReceivedBlockTracker負責管理Spark Streaming運行程序的元數據。數據層面 
2. DStream和JobGenerator是作業調度的核心層面，也就是具體調度到什么程度了，從運行的考慮的。DStream是邏輯層面。 
3. 作業生存層面，JobGenerator是Job調度層面，具體調度到什么程度了。從運行的角度的。

談Driver容錯你要考慮Driver中有那些需要維持狀態的運行。 
1. ReceivedBlockTracker跟蹤了數據，因此需要容錯。通過WAL方式容錯。 
2. DStreamGraph表達了依賴關系，恢復狀態的時候需要根據DStream恢復計算邏輯級別的依賴關系。通過checkpoint方式容錯。 
3. JobGenerator表面你是怎么基于ReceiverBlockTracker中的數據，以及DStream構成的依賴關系不斷的產生Job的過程。你消費了那些數據，進行到什么程度了。

總結如下：

ReceivedBlockTracker： 
1. ReceivedBlockTracker會管理Spark Streaming運行過程中所有的數據。并且把數據分配給需要的batches，所有的動作都會被WAL寫入到Log中，Driver失敗的話，就可以根據歷史恢復tracker狀態，在ReceivedBlockTracker創建的時候，使用checkpoint保存歷史目錄。

下面就從Receiver收到數據之后，怎么處理的開始。 
2. ReceiverBlockTracker.addBlock源碼如下： 
Receiver接收到數據，把元數據信息匯報上來，然后通過ReceiverSupervisorImpl就將數據匯報上來，就直接通過WAL進行容錯. 
當Receiver的管理者，ReceiverSupervisorImpl把元數據信息匯報給Driver的時候，正在處理是交給ReceiverBlockTracker. ReceiverBlockTracker將數據寫進WAL文件中，然后才會寫進內存中，被當前的Spark Streaming程序的調度器使用的，也就是JobGenerator使用的。JobGenerator不可能直接使用WAL。WAL的數據在磁盤中，這里JobGenerator使用的內存中緩存的數據結構

/** Add received block. This event will get written to the write ahead log (if enabled). */
def addBlock(receivedBlockInfo: ReceivedBlockInfo): Boolean = {
try {
val writeResult = writeToLog(BlockAdditionEvent(receivedBlockInfo)) //接收數據后，先進行WAL
if (writeResult) {
      synchronized {
getReceivedBlockQueue(receivedBlockInfo.streamId) += receivedBlockInfo //當WAL成功后，將Block Info元數據信息加入到Block Queue中
      }
      logDebug(s"Stream ${receivedBlockInfo.streamId} received " +
s"block ${receivedBlockInfo.blockStoreResult.blockId}")
    } else {
      logDebug(s"Failed to acknowledge stream ${receivedBlockInfo.streamId} receiving " +
s"block ${receivedBlockInfo.blockStoreResult.blockId} in the Write Ahead Log.")
    }
    writeResult
  } catch {
case NonFatal(e) =>
      logError(s"Error adding block $receivedBlockInfo", e)
false
}
}

Driver端接收到的數據保存在streamIdToUnallocatedBlockQueues中，具體結構如下：

private type ReceivedBlockQueue = mutable.Queue[ReceivedBlockInfo]
private val streamIdToUnallocatedBlockQueues = new mutable.HashMap[Int, ReceivedBlockQueue]

allocateBlocksToBatch把接收到的數據分配給batch，根據streamId取出Block，由此就知道Spark Streaming處理數據的時候可以有不同數據來源

那到底什么是batchTime? 
batchTime是上一個Job分配完數據之后，開始再接收到的數據的時間。

/**
 * Allocate all unallocated blocks to the given batch.
 * This event will get written to the write ahead log (if enabled).
 */
def allocateBlocksToBatch(batchTime: Time): Unit = synchronized {
if (lastAllocatedBatchTime == null || batchTime > lastAllocatedBatchTime) {
val streamIdToBlocks = streamIds.map { streamId =>
        (streamId, getReceivedBlockQueue(streamId).dequeueAll(x => true)) //根據StreamId獲取Block信息
    }.toMap
val allocatedBlocks = AllocatedBlocks(streamIdToBlocks)
if (writeToLog(BatchAllocationEvent(batchTime, allocatedBlocks))) {
timeToAllocatedBlocks.put(batchTime, allocatedBlocks)
lastAllocatedBatchTime = batchTime //這里有對batchTime進行賦值，就是上一個job分配完數據后，開始在接收到數據的時間
    } else {
      logInfo(s"Possibly processed batch $batchTime need to be processed again in WAL recovery")
    }
  } else {
logInfo(s"Possibly processed batch $batchTime need to be processed again in WAL recovery")
  }
}

隨著時間的推移，會不斷產生RDD，這時就需要清理掉一些歷史數據，可以通過cleanupOldBatches方法來清理歷史數據

/**
 * Clean up block information of old batches. If waitForCompletion is true, this method
 * returns only after the files are cleaned up.
 */
def cleanupOldBatches(cleanupThreshTime: Time, waitForCompletion: Boolean): Unit = synchronized {
require(cleanupThreshTime.milliseconds < clock.getTimeMillis())
val timesToCleanup = timeToAllocatedBlocks.keys.filter { _ < cleanupThreshTime }.toSeq
  logInfo("Deleting batches " + timesToCleanup)
if (writeToLog(BatchCleanupEvent(timesToCleanup))) {
timeToAllocatedBlocks --= timesToCleanup
writeAheadLogOption.foreach(_.clean(cleanupThreshTime.milliseconds, waitForCompletion))
  } else {
    logWarning("Failed to acknowledge batch clean up in the Write Ahead Log.")
  }
}

以上幾個方法都進行了WAL動作

(record: ReceivedBlockTrackerLogEvent): = {
(isWriteAheadLogEnabled) {
    logTrace(record)
{
.get.write(ByteBuffer.(Utils.(record))clock.getTimeMillis())
} {
(e) =>
        logWarning(recorde)
}
  } {
}
}

總結： 
WAL對數據的管理包括數據的生成，數據的銷毀和消費。上述在操作之后都要先寫入到WAL的文件中. 

JobGenerator: 
Checkpoint會有時間間隔Batch Duractions，Batch執行前和執行后都會進行checkpoint。 
doCheckpoint被調用的前后流程： 

1、簡單看下generateJobs

/** Generate jobs and perform checkpoint for the given `time`.  */
private def generateJobs(time: Time) {
// Set the SparkEnv in this thread, so that job generation code can access the environment
  // Example: BlockRDDs are created in this thread, and it needs to access BlockManager
  // Update: This is probably redundant after threadlocal stuff in SparkEnv has been removed.
SparkEnv.set(ssc.env)
Try {
    jobScheduler.receiverTracker.allocateBlocksToBatch(time) // allocate received blocks to batch
graph.generateJobs(time) // generate jobs using allocated block
} match {
case Success(jobs) =>
val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
      jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))
case Failure(e) =>
      jobScheduler.reportError("Error generating jobs for time " + time, e)
  }
eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false)) //job完成后就需要進行checkpoint動作
}

2、processEvent接收到消息事件

/** Processes all events */
private def processEvent(event: JobGeneratorEvent) {
  logDebug("Got event " + event)
  event match {
case GenerateJobs(time) => generateJobs(time)
case ClearMetadata(time) => clearMetadata(time)
case DoCheckpoint(time, clearCheckpointDataLater) =>
doCheckpoint(time, clearCheckpointDataLater) // 調用doCheckpoint方法
case ClearCheckpointData(time) => clearCheckpointData(time)
  }
}

3、doCheckpoint源碼如下：

/** Perform checkpoint for the give `time`. */
private def doCheckpoint(time: Time, clearCheckpointDataLater: Boolean) {
if (shouldCheckpoint && (time - graph.zeroTime).isMultipleOf(ssc.checkpointDuration)) {
    logInfo("Checkpointing graph for time " + time)
ssc.graph.updateCheckpointData(time) //最終是進行RDD的Checkpoint
checkpointWriter.write(new Checkpoint(ssc, time), clearCheckpointDataLater)
  }
}

4、DStream中的updateCheckpointData源碼如下：最終導致RDD的Checkpoint

/**
 * Refresh the list of checkpointed RDDs that will be saved along with checkpoint of
 * this stream. This is an internal method that should not be called directly. This is
 * a default implementation that saves only the file names of the checkpointed RDDs to
 * checkpointData. Subclasses of DStream (especially those of InputDStream) may override
 * this method to save custom checkpoint data.
 */
private[streaming] def updateCheckpointData(currentTime: Time) {
  logDebug("Updating checkpoint data for time " + currentTime)
checkpointData.update(currentTime)
  dependencies.foreach(_.updateCheckpointData(currentTime))
  logDebug("Updated checkpoint data for time " + currentTime + ": " + checkpointData)
}

JobGenerator容錯安全性如下圖： 

參考博客：http://blog.csdn.net/snail_gesture/article/details/51492873#comments