Spark Streaming源代码学习总结(一)

1.Spark Streaming 代码分析：

1.1 演示样例代码DEMO：

实时计算的WorldCount：

    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.streaming.StreamingContext._
    import org.apache.spark.storage.StorageLevel

    object NetworkWordCount {
      def main(args: Array[String]) {
        if (args.length < 3) {
          System.err.println("Usage: NetworkWordCount <master> <hostname> <port>\n" +
            "In local mode, <master> should be ‘local[n]‘ with n > 1")
          System.exit(1)
        }

        // Create the context with a 1 second batch size
        val ssc = new StreamingContext("local[4]", "NetworkWordCount", Seconds(1),
          System.getenv("SPARK_HOME"))

        // Create a NetworkInputDStream on target ip:port and count the
        // words in input stream of \n delimited text (eg. generated by ‘nc‘)
        val lines = ssc.socketTextStream(args(1), args(2).toInt, StorageLevel.MEMORY_ONLY_SER)
        val words = lines.flatMap(_.split(" "))
        val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
        wordCounts.print()
        ssc.start()
        ssc.awaitTermination()
      }
    }

以上代码能实时的接受和处理网络传输过来的文本，并计算单词个数。

1.2 以下基于以上代码DEMO分析记录下Spark streaming的启动过程的源代码，以及字节流处理过程的代源代码。

以上代码DEMO涉及的启动流程时序例如以下： 简述：

(1)从Streaming WorldCount的编程演示样例可看出Spark Streaming和Spark的编程API非常像。

Spark Streaming是构建Spark程序框架的基础上的。从时序图的StreamingContext和SparkContext的交互能够看出来StreamingContext里面封装了SparkContext。部分关键代码例如以下：

    //包权限，包内可见。包外不可见
//StreamingContext 里面包装的还是一个SparkContext
class StreamingContext private[streaming] (
    sc_ : SparkContext,
    cp_ : Checkpoint,
    batchDur_ : Duration
  ) extends Logging {

  def this(conf: SparkConf, batchDuration: Duration) = {
    this(StreamingContext.createNewSparkContext(conf), null, batchDuration)
  }

private[streaming] def createNewSparkContext(
    master: String,
    appName: String,
    sparkHome: String,
    jars: Seq[String],
    environment: Map[String, String]
 ): SparkContext = {
    val conf = SparkContext.updatedConf(
     new SparkConf(), master, appName, sparkHome, jars, environment)
    createNewSparkContext(conf)
}
...

上面是StreamingContext的部分构造函数的代码。可见当中对SparkContext的封装以及CreateNewSparkContext对于SparkContext的构造过程。从代码角度看展示了StreamingContext是针对于SparkContext的封装。结合DEMO从框架启动阶段来讲，当完毕StreamingContext的实例化时候也就是完毕了程序的配置初始化过程。

(2)结合Demo，StreamingContext实例化之后则是从StreamingContext获取了一个SocketTextStream，代码例如以下：

val lines = ssc.socketTextStream(args(1), args(2).toInt, StorageLevel.MEMORY_ONLY_SER)

当中args（1）。args（2）分别相应于后面所说的Receiver监听的Ip和port号，随后的參数从英文名能够看出来定义的是存储级别(这里定义仅存储在内存)。

DEMO中的代码简单解释下，然后还是结合时序图和源代码来看下：

StreamingContext.SocketTextStream方法例如以下：

def socketTextStream(
  hostname: String,
  port: Int,
  storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String] = {
    socketStream[String](hostname, port, SocketReceiver.bytesToLines, storageLevel)
}

方法返回的是ReceiverInputDStreaming。这里引出了DStream的概念（拓展总结一下）。依据DStream的代码凝视。DStream的解释例如以下：Dstream本质就是离散化的stream，将stream离散化成一组RDD的list。所以主要的操作仍然是以RDD为基础。

InputDStream 涉及的类继承结构:

上述类继承结构图展示了DInputStream，依据类的名字能够大致看出来DStream的作用，比方FileInputDStream就是处理文件流的DStream，kalfkaInputDSteam就是处理kalfka消息的DStream；DStream本身封装了按时间片离散化了的Stream(数据流）；看例如以下DStream的代码片段：

abstract class DStream[T: ClassTag] (
 @transient private[streaming] var ssc: StreamingContext
  ) extends Serializable with Logging {

  @transient
  private[streaming] var generatedRDDs = new HashMap[Time, RDD[T]] ()
...

如代码所看到的。generatedRDDS是DStream的一个成员变量HashMap，当中key是时间片段。value是RDD。从DStream持有的这个成员变量就能够看出来DStream的本质，就是依照时间片存储了一系列的数据流。

    /**
     * Retrieve a precomputed RDD of this DStream, or computes the RDD. This is an internal
      * method that should not be called directly.
    */
private[streaming] def getOrCompute(time: Time): Option[RDD[T]] = {
    // If this DStream was not initialized (i.e., zeroTime not set), then do it
    // If RDD was already generated, then retrieve it from HashMap
  generatedRDDs.get(time) match {

  // If an RDD was already generated and is being reused, then
  // probably all RDDs in this DStream will be reused and hence should be cached
  case Some(oldRDD) => Some(oldRDD)

  // if RDD was not generated, and if the time is valid
  // (based on sliding time of this DStream), then generate the RDD
  case None => {
    if (isTimeValid(time)) {
      compute(time) match {
        case Some(newRDD) =>
          if (storageLevel != StorageLevel.NONE) {
            newRDD.persist(storageLevel)
            logInfo("Persisting RDD " + newRDD.id + " for time " +
              time + " to " + storageLevel + " at time " + time)
          }
          if (checkpointDuration != null &&
            (time - zeroTime).isMultipleOf(checkpointDuration)) {
            newRDD.checkpoint()
            logInfo("Marking RDD " + newRDD.id + " for time " + time +
              " for checkpointing at time " + time)
          }
          generatedRDDs.put(time, newRDD)
          Some(newRDD)
        case None =>
          None
      }
    } else {
      None
    }
  }
}

以上这段代码则是按时间片构建和检索RDD的主要函数，因为逻辑比較简单就不多说！

DStream不在深入。还是回到正题，上面的DEMO代码从SteamingContext拿到了一个ReceiverInputDStream，ReceiverInputDStream封装了SocketReceiver对象，用来从网络中读取数据流，事实上代码逻辑跟下去，事实上就是拿到了一个SocketInputStream,然后按时间片将流离散化的存储在DStream定义的HashMap中。这段涉及的代码例如以下：

def socketTextStream(
  hostname: String,
  port: Int,
  storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String] = {
socketStream[String](hostname, port, SocketReceiver.bytesToLines, storageLevel)
}
 def socketStream[T: ClassTag](
  hostname: String,
  port: Int,
  converter: (InputStream) => Iterator[T],
  storageLevel: StorageLevel
): ReceiverInputDStream[T] = {
new SocketInputDStream[T](this, hostname, port, converter, storageLevel)
}

以上就是clientDEMO（Driver）获取socketTextStream的过程，总结一下就是通过StreamingContext提供的API获取了一个SocketInputStreamInputDStream。

至此返回到DEMO演示样例。除了WordCount的部分处理逻辑外，就是StreamingContext的启动逻辑了。

通过Spark的仅仅是能够知道RDD仅仅是构建了代码的逻辑以及依赖关系。但并没有真正的运行，同理这里也是相同的道理。以下就是关键的StreamingContext的启动逻辑了。

以下继续总结。

(3)回到时序图，我们能够看出StreamingContext的start方法。直接或间接依次触发了JobScheduler、StreamingListenerBus、ReceiverTracker、JobGenerator的方法。以下结合源代码看一下：

StreamingContext调用了核心的JobScheduler的start方法例如以下：

def start(): Unit = synchronized {
    if (eventActor != null) return // scheduler has already been started

    logDebug("Starting JobScheduler")
    eventActor = ssc.env.actorSystem.actorOf(Props(new Actor {
      def receive = {
        case event: JobSchedulerEvent => processEvent(event)
      }
    }), "JobScheduler")
    /**
     * 启动listenerBus
     */
    listenerBus.start()
    receiverTracker = new ReceiverTracker(ssc)
    //启动receiverTracker
    receiverTracker.start()
    jobGenerator.start()
    logInfo("Started JobScheduler")
}

上面代码简单总结一下：首先生成了一个eventActor对象（类型是Akka Actor; Akka的简单定义：http://baike.baidu.com/view/5912486.htm?fr=aladdin，详细能够深入学下Scala）。eventActor的事件处理逻辑里面调用了processEvent方法。源代码例如以下：

private def processEvent(event: JobSchedulerEvent) {
    try {
      event match {
        case JobStarted(job) => handleJobStart(job)
        case JobCompleted(job) => handleJobCompletion(job)
        case ErrorReported(m, e) => handleError(m, e)
      }
    } catch {
      case e: Throwable =>
        reportError("Error in job scheduler", e)
    }
}
//这个是间接调用 捕获job的启动事件
private def handleJobStart(job: Job) {
    val jobSet = jobSets.get(job.time)
    if (!jobSet.hasStarted) {
      listenerBus.post(StreamingListenerBatchStarted(jobSet.toBatchInfo))
    }
    jobSet.handleJobStart(job)
    logInfo("Starting job " + job.id + " from job set of time " + jobSet.time)
    SparkEnv.set(ssc.env)
}
//捕获job的完毕事件
private def handleJobCompletion(job: Job) {
    job.result match {
      case Success(_) =>
        val jobSet = jobSets.get(job.time)
        jobSet.handleJobCompletion(job)
        logInfo("Finished job " + job.id + " from job set of time " + jobSet.time)
        if (jobSet.hasCompleted) {
          jobSets.remove(jobSet.time)
          jobGenerator.onBatchCompletion(jobSet.time)
          logInfo("Total delay: %.3f s for time %s (execution: %.3f s)".format(
            jobSet.totalDelay / 1000.0, jobSet.time.toString,
            jobSet.processingDelay / 1000.0
          ))
          listenerBus.post(StreamingListenerBatchCompleted(jobSet.toBatchInfo))
        }
      case Failure(e) =>
        reportError("Error running job " + job, e)
    }
}
//捕获job的ErrorReport事件
private def handleError(msg: String, e: Throwable) {
    logError(msg, e)
    ssc.waiter.notifyError(e)
}

至此能够看到JobScheduler作为Job的调度器，所通过Akka框架的eventActor对象能够实时监听到Job的启动，停止以及错误等信息。

而且通过以上列出的几个Handle*方法能够看出来在处理完開始结束时间后。将事件信息做了简单的封装后传输到了ListenerBus，让注冊在监听器总线上的监听器都能针对Job的状态变化及时作出反应。这个ListenerBus的设计还是十分巧妙的。

EventActor启动完成后就启动了StreamingListenerBus对象。

涉及的ListenerBus.start的核心源代码例如以下：

//启动函数
def start() {
    listenerThread.start()
}

//启动一条Daemon线程
val listenerThread = new Thread("StreamingListenerBus") {
setDaemon(true)
override def run() {
  while (true) {
    val event = eventQueue.take
    event match {
      case receiverStarted: StreamingListenerReceiverStarted =>
        listeners.foreach(_.onReceiverStarted(receiverStarted))
      case receiverError: StreamingListenerReceiverError =>
        listeners.foreach(_.onReceiverError(receiverError))
      case receiverStopped: StreamingListenerReceiverStopped =>
        listeners.foreach(_.onReceiverStopped(receiverStopped))
      case batchSubmitted: StreamingListenerBatchSubmitted =>
        listeners.foreach(_.onBatchSubmitted(batchSubmitted))
      case batchStarted: StreamingListenerBatchStarted =>
        listeners.foreach(_.onBatchStarted(batchStarted))
      case batchCompleted: StreamingListenerBatchCompleted =>
        listeners.foreach(_.onBatchCompleted(batchCompleted))
      case StreamingListenerShutdown =>
        // Get out of the while loop and shutdown the daemon thread
        return
      case _ =>
    }
  }
}

通过以上代码能够看出监听器总线真实代码如其名，其start方法启动了一条Deamon线程，来监听各种事件，然后将时间捕捉后发送给注冊在总线上的监听器，其设计的巧妙可见一斑。

以下在没忍住贴上ListenerBus完整的源代码：

/** Asynchronously passes StreamingListenerEvents to registered StreamingListeners. */
private[spark] class StreamingListenerBus() extends Logging {
  private val listeners = new ArrayBuffer[StreamingListener]()
    with SynchronizedBuffer[StreamingListener]

  /* Cap the capacity of the SparkListenerEvent queue so we get an explicit error (rather than
   * an OOM exception) if it‘s perpetually being added to more quickly than it‘s being drained. */
  private val EVENT_QUEUE_CAPACITY = 10000
  private val eventQueue = new LinkedBlockingQueue[StreamingListenerEvent](EVENT_QUEUE_CAPACITY)
  private var queueFullErrorMessageLogged = false

  val listenerThread = new Thread("StreamingListenerBus") {
    setDaemon(true)
    override def run() {
      while (true) {
        val event = eventQueue.take
        event match {
          case receiverStarted: StreamingListenerReceiverStarted =>
            listeners.foreach(_.onReceiverStarted(receiverStarted))
          case receiverError: StreamingListenerReceiverError =>
            listeners.foreach(_.onReceiverError(receiverError))
          case receiverStopped: StreamingListenerReceiverStopped =>
            listeners.foreach(_.onReceiverStopped(receiverStopped))
          case batchSubmitted: StreamingListenerBatchSubmitted =>
            listeners.foreach(_.onBatchSubmitted(batchSubmitted))
          case batchStarted: StreamingListenerBatchStarted =>
            listeners.foreach(_.onBatchStarted(batchStarted))
          case batchCompleted: StreamingListenerBatchCompleted =>
            listeners.foreach(_.onBatchCompleted(batchCompleted))
          case StreamingListenerShutdown =>
            // Get out of the while loop and shutdown the daemon thread
            return
          case _ =>
        }
      }
    }
  }

  def start() {
    listenerThread.start()
  }

  def addListener(listener: StreamingListener) {
    listeners += listener
  }

  def post(event: StreamingListenerEvent) {
    val eventAdded = eventQueue.offer(event)
    if (!eventAdded && !queueFullErrorMessageLogged) {
      logError("Dropping StreamingListenerEvent because no remaining room in event queue. " +
        "This likely means one of the StreamingListeners is too slow and cannot keep up with the " +
        "rate at which events are being started by the scheduler.")
      queueFullErrorMessageLogged = true
    }
  }

  /**
   * Waits until there are no more events in the queue, or until the specified time has elapsed.
   * Used for testing only. Returns true if the queue has emptied and false is the specified time
   * elapsed before the queue emptied.
   */
  def waitUntilEmpty(timeoutMillis: Int): Boolean = {
    val finishTime = System.currentTimeMillis + timeoutMillis
    while (!eventQueue.isEmpty) {
      if (System.currentTimeMillis > finishTime) {
        return false
      }
      /* Sleep rather than using wait/notify, because this is used only for testing and wait/notify
       * add overhead in the general case. */
      Thread.sleep(10)
    }
    true
  }

  def stop(): Unit = post(StreamingListenerShutdown)
}

通过这个监听器总线的完整源代码，能够看到里面一个监听器数组。以及一个事件的堵塞队列，以及事件处理程序，也就是将事件通知到注冊上来的每一个监听器。这段代码这是看着就舒服。是一个非常好的代码模式呀，赞一个！

陶醉至此继续看ReceiverTracker（ReceiverTracker是Receiver监管程序）的启动过程,首先看下ReceiverTracker的start方法：

 /** Start the actor and receiver execution thread. */
def start() = synchronized {
    if (actor != null) {
      throw new SparkException("ReceiverTracker already started")
    }

    if (!receiverInputStreams.isEmpty) {
      actor = ssc.env.actorSystem.actorOf(Props(new ReceiverTrackerActor),
        "ReceiverTracker")
      receiverExecutor.start()
      logInfo("ReceiverTracker started")
    }
}

解读一下首先还是new了一个Akka的Actor对象，肯定是用来处理一些事件呀。看代码：

/** Actor to receive messages from the receivers. */
private class ReceiverTrackerActor extends Actor {
    def receive = {
      case RegisterReceiver(streamId, typ, host, receiverActor) =>
        registerReceiver(streamId, typ, host, receiverActor, sender)
        sender ! true
      case AddBlock(receivedBlockInfo) =>
        addBlocks(receivedBlockInfo)
      case ReportError(streamId, message, error) =>
        reportError(streamId, message, error)
      case DeregisterReceiver(streamId, message, error) =>
        deregisterReceiver(streamId, message, error)
        sender ! true
    }
}

这个Actor主要是监听从Receiver发送过来的一些消息，包含Receiver的注冊、注销，以及接受数据流的存储AddBlock。再次可见JobTracker真是统揽全局的组件。

ReceiverTracker start方法初始化监听Receiver的Actor后，以下的 receiverExecutor.start()方法的调用才是真正的启动Receiver的核心逻辑。看代码：

从这个代码片段能够看出receiverExecutor是ReceiverLaucher的实例。其Start逻辑例如以下：

 def start() {
  thread.start()
}

/** This thread class runs all the receivers on the cluster.  */
class ReceiverLauncher {
    @transient val env = ssc.env
    @transient val thread  = new Thread() {
    override def run() {
    try {
      SparkEnv.set(env)
      //启动Receiver
      startReceivers()
    } catch {
      case ie: InterruptedException => logInfo("ReceiverLauncher interrupted")
    }
  }
}

 /**
 * Get the receivers from the ReceiverInputDStreams, distributes them to the
 * worker nodes as a parallel collection, and runs them.
 */
private def startReceivers() {
  val receivers = receiverInputStreams.map(nis => {
    val rcvr = nis.getReceiver()
    rcvr.setReceiverId(nis.id)
    rcvr
  })

  // Right now, we only honor preferences if all receivers have them
  val hasLocationPreferences = receivers.map(_.preferredLocation.isDefined).reduce(_ && _)

  // Create the parallel collection of receivers to distributed them on the worker nodes
  val tempRDD =
    if (hasLocationPreferences) {
      val receiversWithPreferences = receivers.map(r => (r, Seq(r.preferredLocation.get)))
      ssc.sc.makeRDD[Receiver[_]](receiversWithPreferences)
    } else {
      ssc.sc.makeRDD(receivers, receivers.size)
    }

  // Function to start the receiver on the worker node
  val startReceiver = (iterator: Iterator[Receiver[_]]) => {
    if (!iterator.hasNext) {
      throw new SparkException(
        "Could not start receiver as object not found.")
    }
    val receiver = iterator.next()
    val executor = new ReceiverSupervisorImpl(receiver, SparkEnv.get)
    executor.start()
    executor.awaitTermination()
  }
  // Run the dummy Spark job to ensure that all slaves have registered.
  // This avoids all the receivers to be scheduled on the same node.
  if (!ssc.sparkContext.isLocal) {
    ssc.sparkContext.makeRDD(1 to 50, 50).map(x => (x, 1)).reduceByKey(_ + _, 20).collect()
  }

  // Distribute the receivers and start them
  logInfo("Starting " + receivers.length + " receivers")
  ssc.sparkContext.runJob(tempRDD, startReceiver)
  logInfo("All of the receivers have been terminated")
}

能够ReceiverLauncher的start方法相同启动一条线程，异步的从InputStreams获取全部的Receiver，然后将Receiver封装成ReceiverSuperior逐个启动。当然这么这个是笼统的总结，从细节上再看下:

上面方法中是从receiverInputStreams获取了全部的Receiver，ReceiverInputStreams事实上是这个样子的。

private val inputStreams = new ArrayBuffer[InputDStream[_]]()

def getReceiverInputStreams() = this.synchronized {
inputStreams.filter(_.isInstanceOf[ReceiverInputDStream[_]])
  .map(_.asInstanceOf[ReceiverInputDStream[_]])
  .toArray
 }

一个InputDStream数组经过以下函数处理后得到的数组,主要是校验一下是否是ReceiverInputDStream类型。

从封装在InputDStream中拿到全部的Receiver结合后，然后将Receiver封装成RDD分发到多个work结点上（可见Spark Streaming的巧妙，将Receiver像RDD一样分发）代码例如以下：

  //将Receiver封装成RDD并分发到work结点上。
  val tempRDD =
    if (hasLocationPreferences) {
      val receiversWithPreferences = receivers.map(r => (r, Seq(r.preferredLocation.get)))
      ssc.sc.makeRDD[Receiver[_]](receiversWithPreferences)
    } else {
      ssc.sc.makeRDD(receivers, receivers.size)
  }

  //在worker结点上循环遍历并将Receiver封装成ReceiverSupervisor然后逐个启动。
  val startReceiver = (iterator: Iterator[Receiver[_]]) => {
      if (!iterator.hasNext) {
        throw new SparkException(
          "Could not start receiver as object not found.")
      }
      val receiver = iterator.next()
      val executor = new ReceiverSupervisorImpl(receiver, SparkEnv.get)
      executor.start()
      executor.awaitTermination()
  }

这里有个细节是每一个Receiver事实上都是封装成了ReceiverSupervisor，然后才启动的。ReceiverSupervisor事实上是Receiver的监管程序，其能够用来处理Receiver启动之外的一些事件逻辑，至此补充一张ReceiverTracker以及ReceverSupervisor相关的时序图：

因为Receiver都是封装在ReceiverSupervisor里的。那就看一下ReceiverSupervisor的start方法。

 /** Start the supervisor */
  def start() {
    onStart()
    startReceiver()
  }

  override protected def onStart() {
    blockGenerator.start()
  }
   /** Start receiver */
  def startReceiver(): Unit = synchronized {
    try {
      logInfo("Starting receiver")
      receiver.onStart()
      logInfo("Called receiver onStart")
      onReceiverStart()
      receiverState = Started
    } catch {
      case t: Throwable =>
        stop("Error starting receiver " + streamId, Some(t))
    }
  }

  /**
    * 发送RegisterServer 消息给Driver报告自己启动了。

*/
  override protected def onReceiverStart() {
    val msg = RegisterReceiver(
      streamId, receiver.getClass.getSimpleName, Utils.localHostName(), actor)
    val future = trackerActor.ask(msg)(askTimeout)
    Await.result(future, askTimeout)
  }

结合时序图以及源代码，能够看出ReceiverSupervisor的start方法触发了BlockGenerator的start方法以及Receiver的onstart方法。

那么一下补充一张有关BlockGenerator的相关调用时序以及Receiver的类继承结构图。

SocketReceiver接受并保存数据的处理流程。

Receiver的类继承结构图：

依据以上Receiver的类结构图，能够看出来Receiver的继承堆栈结构，我们仅仅看下和本次Demo相关的onstart方法（由于ReceiverSupervisor调用了Recevier的onstart方法。而且这里涉及的Receiver是SocketReceiver）。

private[streaming]
class SocketReceiver[T: ClassTag](
    host: String,
    port: Int,
    bytesToObjects: InputStream => Iterator[T],
    storageLevel: StorageLevel
  ) extends Receiver[T](storageLevel) with Logging {

  def onStart() {
    // Start the thread that receives data over a connection
    new Thread("Socket Receiver") {
      setDaemon(true)
      override def run() { receive() }
    }.start()
  }

  def onStop() {
    // There is nothing much to do as the thread calling receive()
    // is designed to stop by itself isStopped() returns false
  }

  /** Create a socket connection and receive data until receiver is stopped */
  def receive() {
    var socket: Socket = null
    try {
      logInfo("Connecting to " + host + ":" + port)
      socket = new Socket(host, port)
      logInfo("Connected to " + host + ":" + port)
      val iterator = bytesToObjects(socket.getInputStream())
      while(!isStopped && iterator.hasNext) {
        store(iterator.next)
      }
      logInfo("Stopped receiving")
      restart("Retrying connecting to " + host + ":" + port)
    } catch {
      case e: java.net.ConnectException =>
        restart("Error connecting to " + host + ":" + port, e)
      case t: Throwable =>
        restart("Error receiving data", t)
    } finally {
      if (socket != null) {
        socket.close()
        logInfo("Closed socket to " + host + ":" + port)
      }
    }
  }
}

从以上代码能够看出SocketReceiver的onStart方法也是启动了一条后台线程，循环的监听Socket的还有一端发送过来的字节流，然后调用store方法保存起来，供兴许的Action进行处理。

那就在看下Store方法:

  /**
   * Store a single item of received data to Spark‘s memory.
   * These single items will be aggregated together into data blocks before
   * being pushed into Spark‘s memory.
   */
  def store(dataItem: T) {
    executor.pushSingle(dataItem)
  }

结合上端代码和并结合上面“socketReceiver接受并保存数据”的时序图。就可以追溯到SocketReceiver接受数据并有BlockManager报错的过程。当中ReceiverSupervisor重载了BlockManager的“+”号方法，同一时候也利用了堵塞队列的机制来保存数据。最后通过BlockManager将接受到的流数据存储起来。

以上过程就不做具体分解和结合时序图还是比較easy理解的。

(4)最后单拿出来再看下JobGenerator的start方法：

     /** Start generation of jobs */
    def start(): Unit = synchronized {
        if (eventActor != null) return // generator has already been started

        eventActor = ssc.env.actorSystem.actorOf(Props(new Actor {
          def receive = {
            case event: JobGeneratorEvent =>  processEvent(event)
          }
        }), "JobGenerator")
        if (ssc.isCheckpointPresent) {
          restart()
        } else {
          startFirstTime()
        }
    }

    /** Starts the generator for the first time */
   private def startFirstTime() {
    val startTime = new Time(timer.getStartTime())
    graph.start(startTime - graph.batchDuration)
    timer.start(startTime.milliseconds)
    logInfo("Started JobGenerator at " + startTime)
   }

   /** Restarts the generator based on the information in checkpoint */
   private def restart() {
    // If manual clock is being used for testing, then
    // either set the manual clock to the last checkpointed time,
    // or if the property is defined set it to that time
    if (clock.isInstanceOf[ManualClock]) {
      val lastTime = ssc.initialCheckpoint.checkpointTime.milliseconds
      val jumpTime = ssc.sc.conf.getLong("spark.streaming.manualClock.jump", 0)
      clock.asInstanceOf[ManualClock].setTime(lastTime + jumpTime)
    }

    val batchDuration = ssc.graph.batchDuration

    // Batches when the master was down, that is,
    // between the checkpoint and current restart time
    val checkpointTime = ssc.initialCheckpoint.checkpointTime
    val restartTime = new Time(timer.getRestartTime(graph.zeroTime.milliseconds))
    val downTimes = checkpointTime.until(restartTime, batchDuration)
    logInfo("Batches during down time (" + downTimes.size + " batches): "
      + downTimes.mkString(", "))

    // Batches that were unprocessed before failure
    val pendingTimes = ssc.initialCheckpoint.pendingTimes.sorted(Time.ordering)
    logInfo("Batches pending processing (" + pendingTimes.size + " batches): " +
      pendingTimes.mkString(", "))
    // Reschedule jobs for these times
    val timesToReschedule = (pendingTimes ++ downTimes).distinct.sorted(Time.ordering)
    logInfo("Batches to reschedule (" + timesToReschedule.size + " batches): " +
      timesToReschedule.mkString(", "))
    timesToReschedule.foreach(time =>
      jobScheduler.submitJobSet(JobSet(time, graph.generateJobs(time)))
    )

    // Restart the timer
    timer.start(restartTime.milliseconds)
    logInfo("Restarted JobGenerator at " + restartTime)
  }

以上代码则是设计了Job的生成以及启动过程，详细信息临时做个checkPoint有待下次分解。