(版本定制)第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