SparkStreaming源碼之JobScheduler篇

首先看下JobScheduler這個(gè)類是在什么時(shí)候被實(shí)例化的，打開(kāi)StreamingContext代碼可見(jiàn)：

  private[streaming] val scheduler = new JobScheduler(this)

  private[streaming] val waiter = new ContextWaiter

  private[streaming] val progressListener = new StreamingJobProgressListener(this)

再看下job的產(chǎn)生者jobGenerator是如何將生成的job傳遞給JobScheduler的

  /** 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)
        //todo 將生成的job提交給jobScheduler
        jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))
      case Failure(e) =>
        jobScheduler.reportError("Error generating jobs for time " + time, e)
    }
    eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
  }

JobScheduler處理提交上來(lái)的job，并將job存放在jobSet的數(shù)據(jù)結(jié)構(gòu)中

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

SparkStreaming在一個(gè)Application中能夠同時(shí)運(yùn)行多個(gè)job的，其實(shí)就是使用多線程來(lái)實(shí)現(xiàn)

  //todo jobSet數(shù)據(jù)結(jié)構(gòu)
  private val jobSets: java.util.Map[Time, JobSet] = new ConcurrentHashMap[Time, JobSet]
  private val numConcurrentJobs = ssc.conf.getInt("spark.streaming.concurrentJobs", 1)
  //todo 使用線程池來(lái)運(yùn)行多個(gè)job事件
  private val jobExecutor =
    ThreadUtils.newDaemonFixedThreadPool(numConcurrentJobs, "streaming-job-executor"

JobScheduler負(fù)責(zé)job的調(diào)度，在內(nèi)部是使用一個(gè)消息循環(huán)體來(lái)處理job的各種事件，而這個(gè)消息循環(huán)體也是在JobSchduler的start方法中實(shí)例化

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

    //todo 內(nèi)部的消息循環(huán)體
    logDebug("Starting JobScheduler")
    eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") {
      override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event)

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

    // attach rate controllers of input streams to receive batch completion updates
    for {
      inputDStream <- ssc.graph.getInputStreams
      rateController <- inputDStream.rateController
    } ssc.addStreamingListener(rateController)

    listenerBus.start(ssc.sparkContext)
    receiverTracker = new ReceiverTracker(ssc)
    inputInfoTracker = new InputInfoTracker(ssc)
    receiverTracker.start()
    jobGenerator.start()
    logInfo("Started JobScheduler")
  }

看下這個(gè)消息循環(huán)體具體的內(nèi)容，可見(jiàn)Job的啟動(dòng)，完成，還有錯(cuò)誤處理都在這里，具體方法可以點(diǎn)進(jìn)去看

  private def processEvent(event: JobSchedulerEvent) {
    try {
      event match {
        case JobStarted(job, startTime) => handleJobStart(job, startTime)//todo 啟動(dòng)job事件處理
        case JobCompleted(job, completedTime) => handleJobCompletion(job, completedTime)//todo job完成事件處理
        case ErrorReported(m, e) => handleError(m, e)//todo 異常事件處理
      }
    } catch {
      case e: Throwable =>
        reportError("Error in job scheduler", e)
    }
  }

現(xiàn)在看下一個(gè)job的啟動(dòng)，在SubmitJobSet方法，JobExecutor線程池去執(zhí)行每個(gè)JobHandler

  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)
      //todo 在這里處理jobSet里面的每個(gè)job
      jobSet.jobs.foreach(job => jobExecutor.execute(new JobHandler(job)))
      logInfo("Added jobs for time " + jobSet.time)
    }
  }

看下jobHandler這個(gè)線程的run方法

  private class JobHandler(job: Job) extends Runnable with Logging {
    import JobScheduler._

    def run() {
      try {
        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)

        // 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) {
          //todo 這里給自己發(fā)消息啟動(dòng)job，其實(shí)就是打出一些日志
          _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.
          PairRDDFunctions.disableOutputSpecValidation.withValue(true) {
            //todo 這里是關(guān)鍵 一個(gè)job內(nèi)容的運(yùn)行
            job.run()
          }
          _eventLoop = eventLoop
          if (_eventLoop != null) {
            //todo 這里給自己發(fā)消息jo完成，其實(shí)也是打出一些日志
            _eventLoop.post(JobCompleted(job, clock.getTimeMillis()))
          }
        } else {
          // JobScheduler has been stopped.
        }
      } finally {
        ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, null)
        ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, null)
      }
    }
  }
}

看下最終的run方法,這個(gè)run方法執(zhí)行的是job的輸出代碼的方法，例如print操作產(chǎn)生的job

private[streaming]
class Job(val time: Time, func: () => _) {
  private var _id: String = _
  private var _outputOpId: Int = _
  private var isSet = false
  private var _result: Try[_] = null
  private var _callSite: CallSite = null
  private var _startTime: Option[Long] = None
  private var _endTime: Option[Long] = None

  def run() {
    //todo 這里的func()便是你的action操作方法,或者是你傳入的輸入方法
    _result = Try(func())
  }
  
  //todo print操作
  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("Time: " + time)
        println("-------------------------------------------")
        firstNum.take(num).foreach(println)
        if (firstNum.length > num) println("...")
        println()
        // scalastyle:on println
      }
    }
    foreachRDD(context.sparkContext.clean(foreachFunc), displayInnerRDDOps = false)
  }  

至此 JobScheduler角色的工作以敘述完畢！