—————?—————?—————?—————?—————?—————
Spark Streaming概述
Spark Streaming 初始化過程
Spark Streaming Receiver啟動過程分析
Spark Streaming 數(shù)據(jù)準備階段分析（Receiver方式）
Spark Streaming 數(shù)據(jù)計算階段分析
SparkStreaming Backpressure分析
Spark Streaming Executor DynamicAllocation 機制分析

—————?—————?—————?—————?—————?—————

SparkStreaming的全過程分為兩個階段：數(shù)據(jù)準備階段和數(shù)據(jù)計算階段。兩個階段在功能上相互獨立，僅通過數(shù)據(jù)聯(lián)系在一起。"Spark Streaming 數(shù)據(jù)準備階段分析"從源碼角度分析了Spark Streaming數(shù)據(jù)準備階段的具體流程。本文將從源碼的角度對數(shù)據(jù)計算階段的具體流程進行分析。
Spark Streaming數(shù)據(jù)計算階段包含批次數(shù)據(jù)劃分，批作業(yè)生成，批wt提交三個部分。

1、 JobGenerator 啟動

JobGenerator用于定期生成Job并進行提交 。"Spark Streaming 初始化過程分析"中提到，在啟動JobScheduler時，其會調(diào)用JobGenerator的start方法，啟動JobGenerator.
JobGenerator的start方法實現(xiàn)如下：

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

    // Call checkpointWriter here to initialize it before eventLoop uses it to avoid a deadlock.
    // See SPARK-10125
    checkpointWriter

    eventLoop = new EventLoop[JobGeneratorEvent]("JobGenerator") {
      override protected def onReceive(event: JobGeneratorEvent): Unit = processEvent(event)

      override protected def onError(e: Throwable): Unit = {
        jobScheduler.reportError("Error in job generator", e)
      }
    }
    eventLoop.start()

    if (ssc.isCheckpointPresent) {
      restart()
    } else {
      startFirstTime()
    }
  }

通過分析上述代碼可知，在JobGenerator.start()被調(diào)用時，其將創(chuàng)建

• eventLoop對象并啟動，其中eventLoop定義事件交由processEvent(event).
  processEvent其依據(jù)事件的類型，對其進行不同的處理。
• 調(diào)用startFirstTime（）方法。通過分析startFirstTime的實現(xiàn)邏輯，可知其進行兩項主要工作：
• 調(diào)用 timer.start方法、 定期生成Job
• 調(diào)用graph.start方法

  /** 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)
  }

下面我分別對這二者進行解析。

2、Job 生成及提交

2.1 周期性觸發(fā)Job生成事件

startFirstTime()方法中調(diào)用了timer.start方法，其中timer[RecurringTimer]為定時器，與Spark Streaming 數(shù)據(jù)準備階段分析一文中介紹切片時所有定時器一樣。其按設(shè)置的時間周期，重復(fù)的執(zhí)行計劃的任務(wù)。此處Timer的具體實現(xiàn)為：

  private val timer = new RecurringTimer(clock, ssc.graph.batchDuration.milliseconds,
    longTime => eventLoop.post(GenerateJobs(new Time(longTime))), "JobGenerator")

其每個batchDuration規(guī)定時間，都會向eventLoop發(fā)送一GenerateJobs事件，eventLoop收到GenerateJobs事件，則使用processEvent進行相應(yīng)處理，此處為調(diào)用 generateJobs()方法 ,生成job.

  /** 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)
      case ClearCheckpointData(time) => clearCheckpointData(time)
    }
  }

2.2 Job生成詳細過程

上文搞到Timer分周期性的觸發(fā)Job生成事件，并通過generateJobs來生成Job.
JobGenerator在每個Batch Interval都會為應(yīng)用中的每個Output Stream建立一個Job, 該批次中的所有Job組成一個JobSet.使用JobScheduler的submitJobSet進行批量Job提交。
下面來分析generateJobs的實現(xiàn)邏輯。

  /** Generate jobs and perform checkpointing for the given `time`.  */
  private def generateJobs(time: Time) {
    // Checkpoint all RDDs marked for checkpointing to ensure their lineages are
    // truncated periodically. Otherwise, we may run into stack overflows (SPARK-6847).
    ssc.sparkContext.setLocalProperty(RDD.CHECKPOINT_ALL_MARKED_ANCESTORS, "true")
    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)
        PythonDStream.stopStreamingContextIfPythonProcessIsDead(e)
    }
    eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
  }

在generateJobs生成Job時， 其首先通過ReceiverTracker 取得其中注冊的未分配的數(shù)據(jù)信息。然后通過DStreamGraph生成Job。

2.2.1 批數(shù)據(jù)信息劃分

本部分會將Spark Streaming 數(shù)據(jù)準備階段分析 中生成的未分配的block，劃歸到某個批次進行處理。具體過程如下：

在生成Job時，首先調(diào)用如下語句：

jobScheduler.receiverTracker.allocateBlocksToBatch(time)

該語句用來劃分某批次(time)要處理的數(shù)據(jù)。下面對其獲取過程進行詳說說明。

其中allocateBlocksToBatch的實現(xiàn)如下：

  /** Allocate all unallocated blocks to the given batch. */
  def allocateBlocksToBatch(batchTime: Time): Unit = {
    if (receiverInputStreams.nonEmpty) {
      receivedBlockTracker.allocateBlocksToBatch(batchTime)
    }
  }

其將調(diào)用receivedBlockTracker的allocateBlocksToBatch方法，將未分配數(shù)據(jù)信息取出，并劃分給batchTime所指批次。首先receivedBlockTracker從streamIdToUnallocatedBlockQueues中取出未分配的block信息，將其包裝為AllocatedBlocks，并注冊在timeToAllocatedBlocks表中，等待某批次（batchTime）生成Job時，與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))
      }.toMap
      val allocatedBlocks = AllocatedBlocks(streamIdToBlocks)
      if (writeToLog(BatchAllocationEvent(batchTime, allocatedBlocks))) {
        timeToAllocatedBlocks.put(batchTime, allocatedBlocks)
        lastAllocatedBatchTime = batchTime
      } else {
        logInfo(s"Possibly processed batch $batchTime needs to be processed again in WAL recovery")
      }
    } else {
      // This situation occurs when:
      // 1. WAL is ended with BatchAllocationEvent, but without BatchCleanupEvent,
      // possibly processed batch job or half-processed batch job need to be processed again,
      // so the batchTime will be equal to lastAllocatedBatchTime.
      // 2. Slow checkpointing makes recovered batch time older than WAL recovered
      // lastAllocatedBatchTime.
      // This situation will only occurs in recovery time.
      logInfo(s"Possibly processed batch $batchTime needs to be processed again in WAL recovery")
    }
  }

2.2.2 批作業(yè)（Job）生成

通過graph.generateJobs(time)方法分別將DStreamGraph中的每個OutputStream轉(zhuǎn)換了一個Job(如果應(yīng)用中有多個OutputStream算子，則一個批次會生成多個Job)。generateJobs實現(xiàn)邏輯如下：

def generateJobs(time: Time): Seq[Job] = {
    logDebug("Generating jobs for time " + time)
    val jobs = this.synchronized {
      outputStreams.flatMap { outputStream =>
        val jobOption = outputStream.generateJob(time)
        jobOption.foreach(_.setCallSite(outputStream.creationSite))
        jobOption
      }
    }
    logDebug("Generated " + jobs.length + " jobs for time " + time)
    jobs
  }

通過分析源碼，其將逐個調(diào)用OutputStream的generateJob方法來將每個OutputStream轉(zhuǎn)化為Job. OutputStream不同于其它DStream的地方為其重寫了generateJob方法, 以WordCount程序中使用的print算子中的ForEachDStream為例，其 generateJob實現(xiàn)如下：

override def generateJob(time: Time): Option[Job] = {
    parent.getOrCompute(time) match {
      case Some(rdd) =>
        val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
          foreachFunc(rdd, time)
        }
        Some(new Job(time, jobFunc))
      case None => None
    }
  }

通過程序，可以看出，其將調(diào)用父DStream中的getOrCompute方法，生成RDD，然后包裝成Job。

2.2.2.1 RDD 生成

以WordCount為例，先來看一下WordCount應(yīng)用中DStream的轉(zhuǎn)換，轉(zhuǎn)換關(guān)系如下：

WordCount應(yīng)用中DStream轉(zhuǎn)換關(guān)系

通過分析，getOrCompute( compute方法與之類似)方法由DStream基類創(chuàng)建, 如果子類重寫該方法,則執(zhí)行子類方法; 若未重寫,則執(zhí)行基類中的方法。通過查看上述轉(zhuǎn)換關(guān)系鏈中ForEachDStream的父親-ShuffledDStream，發(fā)現(xiàn)其未重寫getOrCompute方法，因此將使用繼承自基類DStream中的getOrCompute， 代碼如下。

  /**
   * Get the RDD corresponding to the given time; either retrieve it from cache
   * or compute-and-cache it.
   */
  private[streaming] final def getOrCompute(time: Time): Option[RDD[T]] = {
    // If RDD was already generated, then retrieve it from HashMap,
    // or else compute the RDD
    generatedRDDs.get(time).orElse {
      // Compute the RDD if time is valid (e.g. correct time in a sliding window)
      // of RDD generation, else generate nothing.
      if (isTimeValid(time)) {

        val rddOption = createRDDWithLocalProperties(time, displayInnerRDDOps = false) {
          // Disable checks for existing output directories in jobs launched by the streaming
          // scheduler, since we may need to write output to an existing directory during checkpoint
          // recovery; see SPARK-4835 for more details. We need to have this call here because
          // compute() might cause Spark jobs to be launched.
          SparkHadoopWriterUtils.disableOutputSpecValidation.withValue(true) {
            compute(time)
          }
        }

        rddOption.foreach { case newRDD =>
          // Register the generated RDD for caching and checkpointing
          if (storageLevel != StorageLevel.NONE) {
            newRDD.persist(storageLevel)
            logDebug(s"Persisting RDD ${newRDD.id} for time $time to $storageLevel")
          }
          if (checkpointDuration != null && (time - zeroTime).isMultipleOf(checkpointDuration)) {
            newRDD.checkpoint()
            logInfo(s"Marking RDD ${newRDD.id} for time $time for checkpointing")
          }
          generatedRDDs.put(time, newRDD)
        }
        rddOption
      } else {
        None
      }
    }
  }

此代碼將調(diào)用ShuffledDStream的compute生成RDD，其compute實現(xiàn)為：

override def compute(validTime: Time): Option[RDD[(K, C)]] = {
    parent.getOrCompute(validTime) match {
      case Some(rdd) => Some(rdd.combineByKey[C](
          createCombiner, mergeValue, mergeCombiner, partitioner, mapSideCombine))
      case None => None
    }
  }

通過分析，其將調(diào)用其父DStream的compute方法，其父DStream繼續(xù)遞歸向上調(diào)用父DStream的compute直到源頭DStream（SocketInputDStream），
SocketInputDStream的compute方法繼承自ReceiverInputDStream，其compute方法將生成源頭RDD，并按DStream遞歸逆向生成RDD Graph.

ReceiverInputDStream定義的compute的實現(xiàn)如下：

 /**
   * Generates RDDs with blocks received by the receiver of this stream. */
  override def compute(validTime: Time): Option[RDD[T]] = {
    val blockRDD = {

      if (validTime < graph.startTime) {
        // If this is called for any time before the start time of the context,
        // then this returns an empty RDD. This may happen when recovering from a
        // driver failure without any write ahead log to recover pre-failure data.
        new BlockRDD[T](ssc.sc, Array.empty)
      } else {
        // Otherwise, ask the tracker for all the blocks that have been allocated to this stream
        // for this batch
        val receiverTracker = ssc.scheduler.receiverTracker
        val blockInfos = receiverTracker.getBlocksOfBatch(validTime).getOrElse(id, Seq.empty)

        // Register the input blocks information into InputInfoTracker
        val inputInfo = StreamInputInfo(id, blockInfos.flatMap(_.numRecords).sum)
        ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)

        // Create the BlockRDD
        createBlockRDD(validTime, blockInfos)
      }
    }
    Some(blockRDD)
  }

此處，通過如下邏輯

val receiverTracker = ssc.scheduler.receiverTracker 
val blockInfos = receiverTracker.getBlocksOfBatch(validTime).getOrElse(id, Seq.empty)

將 2.2.1 節(jié)中提到的劃分過批次的數(shù)據(jù)信息（blockInfos)取出，包裝成StreamInputInfo，然后通過createBlockRDD方法生成RDD. 此處，如果blockInfos信息不空，則生成正常的RDD；若blockInfos為空，則沒有Block的空RDD（new BlockRDD(ssc.sc, Array.empty)）。

2.2.3 Job 的提交

當成功轉(zhuǎn)化為Job之后，然后通過JobScheduler對JobSet進行提交。

case Success(jobs) =>
        val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
        jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))

其中submitJobSet方法實現(xiàn)如下：

  def submitJobSet(jobSet: JobSet) {
    if (jobSet.jobs.isEmpty) {
      logInfo("No jobs added for time " + jobSet.time)
    } else {
      listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))
      jobSets.put(jobSet.time, jobSet)
      jobSet.jobs.foreach(job => jobExecutor.execute(new JobHandler(job)))
      logInfo("Added jobs for time " + jobSet.time)
    }
  }

• 首先將JobSet加入JobSets表中，以便監(jiān)控系統(tǒng)可以追蹤。
• 將Job通過JobHandler進行包裝，然后由ThreadPoolExecutor的execute增加到其workQueue中，等待被調(diào)度執(zhí)行。如果線程池有空閑線程，則其將被調(diào)度。（此部分為Java并發(fā)編程中Executor的相關(guān)內(nèi)容。）
  其中線程池的定義如下所示：

  private val numConcurrentJobs = ssc.conf.getInt("spark.streaming.concurrentJobs", 1)
  private val jobExecutor =
    ThreadUtils.newDaemonFixedThreadPool(numConcurrentJobs, "streaming-job-executor")

通過分析代碼可知，JobScheduler創(chuàng)建一固定長度的daemon線程池jobExecutor ，大小由“spark.streaming.concurrentJobs”，默認為1。 線程池中有多個線程則可以同時執(zhí)行多少個Job， 默認情況下每次只能提交一個Job。當Job來不及執(zhí)行時，會產(chǎn)生堆集，堆集的Job會保存在ThreadPoolExecutor中的workQueue隊列中，等待有空閑線程時被調(diào)度。

• JobHandler是ThreadPoolExecutor中Executor運行的主要任務(wù)，其功能是對提交的Job進行處理，實現(xiàn)如下, 其將通過EventLoop對Job狀態(tài)進行管理，并通過調(diào)用job.run方法，使用Job開始運行。

    def run() {
      val oldProps = ssc.sparkContext.getLocalProperties
      try {
        ssc.sparkContext.setLocalProperties(SerializationUtils.clone(ssc.savedProperties.get()))
        val formattedTime = UIUtils.formatBatchTime(
          job.time.milliseconds, ssc.graph.batchDuration.milliseconds, showYYYYMMSS = false)
        val batchUrl = s"/streaming/batch/?id=${job.time.milliseconds}"
        val batchLinkText = s"[output operation ${job.outputOpId}, batch time ${formattedTime}]"

        ssc.sc.setJobDescription(
          s"""Streaming job from <a href="$batchUrl">$batchLinkText</a>""")
        ssc.sc.setLocalProperty(BATCH_TIME_PROPERTY_KEY, job.time.milliseconds.toString)
        ssc.sc.setLocalProperty(OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString)
        // Checkpoint all RDDs marked for checkpointing to ensure their lineages are
        // truncated periodically. Otherwise, we may run into stack overflows (SPARK-6847).
        ssc.sparkContext.setLocalProperty(RDD.CHECKPOINT_ALL_MARKED_ANCESTORS, "true")

        // We need to assign `eventLoop` to a temp variable. Otherwise, because
        // `JobScheduler.stop(false)` may set `eventLoop` to null when this method is running, then
        // it's possible that when `post` is called, `eventLoop` happens to null.
        var _eventLoop = eventLoop
        if (_eventLoop != null) {
          _eventLoop.post(JobStarted(job, clock.getTimeMillis()))
          // Disable checks for existing output directories in jobs launched by the streaming
          // scheduler, since we may need to write output to an existing directory during checkpoint
          // recovery; see SPARK-4835 for more details.
          SparkHadoopWriterUtils.disableOutputSpecValidation.withValue(true) {
            job.run()
          }
          _eventLoop = eventLoop
          if (_eventLoop != null) {
            _eventLoop.post(JobCompleted(job, clock.getTimeMillis()))
          }
        } else {
          // JobScheduler has been stopped.
        }
      } finally {
        ssc.sparkContext.setLocalProperties(oldProps)
      }
    }

其中Job.run方法，實現(xiàn)如下：

 def run() {
    _result = Try(func())
  }

其將執(zhí)行創(chuàng)建Job時的方法func。WordCount應(yīng)用是ForEachDStream中進行Job創(chuàng)建。其創(chuàng)建方法 上文已經(jīng)提到：

override def generateJob(time: Time): Option[Job] = {
    parent.getOrCompute(time) match {
      case Some(rdd) =>
        val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
          foreachFunc(rdd, time)
        }
        Some(new Job(time, jobFunc))
      case None => None
    }
  }

分析代碼可知，其將調(diào)用foreachFunc, 該方法是創(chuàng)建ForEachDStream時引入的參數(shù)，由print方法定義

/**
   * Print the first num elements of each RDD generated in this DStream. This is an output
   * operator, so this DStream will be registered as an output stream and there materialized.
   */
  def print(num: Int): Unit = ssc.withScope {
    def foreachFunc: (RDD[T], Time) => Unit = {
      (rdd: RDD[T], time: Time) => {
        val firstNum = rdd.take(num + 1)
        // scalastyle:off println
        println("-------------------------------------------")
        println(s"Time: $time")
        println("-------------------------------------------")
        firstNum.take(num).foreach(println)
        if (firstNum.length > num) println("...")
        println()
        // scalastyle:on println
      }
    }
    foreachRDD(context.sparkContext.clean(foreachFunc), displayInnerRDDOps = false)
  }

其中，會調(diào)用rdd.take（）算子， take算子屬于action算子，會觸發(fā)SparkJob的提交，接下來的處理流程與spark 批處理相同。
前述生成的Job，只是Streaming中定義的抽象，與SparkJob（真正進行調(diào)度，生成Task）不同。