Spark源码分析之二:Job的调度模型与运行反馈

简介:         在《Spark源码分析之Job提交运行总流程概述》一文中,我们提到了,Job提交与运行的第一阶段Stage划分与提交,可以分为三个阶段:         1、Job的调度模型与运行反馈;         2、Stage划分;         3、Stage提交:对应TaskSet的生成。

        在《Spark源码分析之Job提交运行总流程概述》一文中,我们提到了,Job提交与运行的第一阶段Stage划分与提交,可以分为三个阶段:

        1、Job的调度模型与运行反馈;

        2、Stage划分;

        3、Stage提交:对应TaskSet的生成。

        今天,我们就结合源码来分析下第一个小阶段:Job的调度模型与运行反馈。


        首先由DAGScheduler负责将Job提交到事件队列eventProcessLoop中,等待调度执行。入口方法为DAGScheduler的runJon()方法。代码如下:

/**
   * Run an action job on the given RDD and pass all the results to the resultHandler function as
   * they arrive.
   *
   * @param rdd target RDD to run tasks on
   * @param func a function to run on each partition of the RDD
   * @param partitions set of partitions to run on; some jobs may not want to compute on all
   *   partitions of the target RDD, e.g. for operations like first()
   * @param callSite where in the user program this job was called
   * @param resultHandler callback to pass each result to
   * @param properties scheduler properties to attach to this job, e.g. fair scheduler pool name
   *
   * @throws Exception when the job fails
   */
  def runJob[T, U](
      rdd: RDD[T],
      func: (TaskContext, Iterator[T]) => U,
      partitions: Seq[Int],
      callSite: CallSite,
      resultHandler: (Int, U) => Unit,
      properties: Properties): Unit = {
      
    // 开始时间
    val start = System.nanoTime
    
    // 调用submitJob()方法,提交Job,返回JobWaiter
    // rdd为最后一个rdd,即target RDD to run tasks on
    // func为该rdd上每个分区需要执行的函数,a function to run on each partition of the RDD
    // partitions为该rdd上需要执行操作的分区集合,set of partitions to run on
    // callSite为用户程序job被调用的地方,where in the user program this job was called
    val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
    
    // JobWaiter调用awaitResult()方法等待结果
    waiter.awaitResult() match {
      case JobSucceeded => // Job运行成功
        logInfo("Job %d finished: %s, took %f s".format
          (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
      case JobFailed(exception: Exception) =>// Job运行失败
        logInfo("Job %d failed: %s, took %f s".format
          (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
        // SPARK-8644: Include user stack trace in exceptions coming from DAGScheduler.
        val callerStackTrace = Thread.currentThread().getStackTrace.tail
        exception.setStackTrace(exception.getStackTrace ++ callerStackTrace)
        throw exception
    }
  }
        runJob()方法就做了三件事:

        首先,获取开始时间,方便最后计算Job执行时间;

        其次,调用submitJob()方法,提交Job,返回JobWaiter类型的对象waiter;

        最后,waiter调用JobWaiter的awaitResult()方法等待Job运行结果,这个运行结果就俩:JobSucceeded代表成功,JobFailed代表失败。

        awaitResult()方法通过轮询标志位_jobFinished,如果为false,则调用this.wait()继续等待,否则说明Job运行完成,返回JobResult,其代码如下:

def awaitResult(): JobResult = synchronized {
    
    // 循环,如果标志位_jobFinished为false,则一直循环,否则退出,返回JobResult
    while (!_jobFinished) {
      this.wait()
    }
    return jobResult
  }
        而这个标志位_jobFinished是在Task运行完成后,如果已完成Task数目等于总Task数目时,或者整个Job运行失败时设置的,随着标志位的设置,Job运行结果jobResult也同步进行设置,代码如下:

// 任务运行完成
  override def taskSucceeded(index: Int, result: Any): Unit = synchronized {
    if (_jobFinished) {
      throw new UnsupportedOperationException("taskSucceeded() called on a finished JobWaiter")
    }
    resultHandler(index, result.asInstanceOf[T])
    finishedTasks += 1
    // 已完成Task数目是否等于总Task数目
    if (finishedTasks == totalTasks) {
      // 设置标志位_jobFinished为ture
      _jobFinished = true
      // 作业运行结果为成功
      jobResult = JobSucceeded
      this.notifyAll()
    }
  }

  // 作业失败
  override def jobFailed(exception: Exception): Unit = synchronized {
    // 设置标志位_jobFinished为ture
    _jobFinished = true
    // 作业运行结果为失败
    jobResult = JobFailed(exception)
    this.notifyAll()
  }


        接下来,看看submitJob()方法,代码定义如下:

/**
   * Submit an action job to the scheduler.
   *
   * @param rdd target RDD to run tasks on
   * @param func a function to run on each partition of the RDD
   * @param partitions set of partitions to run on; some jobs may not want to compute on all
   *   partitions of the target RDD, e.g. for operations like first()
   * @param callSite where in the user program this job was called
   * @param resultHandler callback to pass each result to
   * @param properties scheduler properties to attach to this job, e.g. fair scheduler pool name
   *
   * @return a JobWaiter object that can be used to block until the job finishes executing
   *         or can be used to cancel the job.
   *
   * @throws IllegalArgumentException when partitions ids are illegal
   */
  def submitJob[T, U](
      rdd: RDD[T],
      func: (TaskContext, Iterator[T]) => U,
      partitions: Seq[Int],
      callSite: CallSite,
      resultHandler: (Int, U) => Unit,
      properties: Properties): JobWaiter[U] = {
    
    // Check to make sure we are not launching a task on a partition that does not exist.
    // 检测rdd分区以确保我们不会在一个不存在的partition上launch一个task
    val maxPartitions = rdd.partitions.length
    partitions.find(p => p >= maxPartitions || p < 0).foreach { p =>
      throw new IllegalArgumentException(
        "Attempting to access a non-existent partition: " + p + ". " +
          "Total number of partitions: " + maxPartitions)
    }

    // 为Job生成一个jobId,jobId为AtomicInteger类型,getAndIncrement()确保了原子操作性,每次生成后都自增
    val jobId = nextJobId.getAndIncrement()
    
    // 如果partitions大小为0,即没有需要执行任务的分区,快速返回
    if (partitions.size == 0) {
      // Return immediately if the job is running 0 tasks
      return new JobWaiter[U](this, jobId, 0, resultHandler)
    }

    assert(partitions.size > 0)
    
    // func转化下,否则JobSubmitted无法接受这个func参数,T转变为_
    val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
    
    // 创建一个JobWaiter对象
    val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
    
    // eventProcessLoop加入一个JobSubmitted事件到事件队列中
    eventProcessLoop.post(JobSubmitted(
      jobId, rdd, func2, partitions.toArray, callSite, waiter,
      SerializationUtils.clone(properties)))
    
    // 返回JobWaiter
    waiter
  }
        submitJob()方法一共做了5件事情:

        第一,数据检测,检测rdd分区以确保我们不会在一个不存在的partition上launch一个task,并且,如果partitions大小为0,即没有需要执行任务的分区,快速返回;

        第二,为Job生成一个jobId,该jobId为AtomicInteger类型,getAndIncrement()确保了原子操作性,每次生成后都自增;

        第三,将func转化下,否则JobSubmitted无法接受这个func参数,T转变为_;

        第四,创建一个JobWaiter对象waiter,该对象会在方法结束时返回给上层方法,以用来监测Job运行结果;

        第五,将一个JobSubmitted事件加入到事件队列eventProcessLoop中,等待工作线程轮询调度(速度很快)。

        这里,我们有必要研究下事件队列eventProcessLoop,eventProcessLoop为DAGSchedulerEventProcessLoop类型的,在DAGScheduler初始化时被定义并赋值,代码如下:

// 创建DAGSchedulerEventProcessLoop类型的成员变量eventProcessLoop
  private[scheduler] val eventProcessLoop = new DAGSchedulerEventProcessLoop(this)
        DAGSchedulerEventProcessLoop继承自EventLoop,我们先来看看这个EventLoop的定义。

/**
 * An event loop to receive events from the caller and process all events in the event thread. It
 * will start an exclusive event thread to process all events.
 * EventLoop用来接收来自调用者的事件并在event thread中除了所有的事件。它将开启一个专门的事件处理线程处理所有的事件。
 *
 * Note: The event queue will grow indefinitely. So subclasses should make sure `onReceive` can
 * handle events in time to avoid the potential OOM.
 */
private[spark] abstract class EventLoop[E](name: String) extends Logging {
  
  // LinkedBlockingDeque类型的事件队列,队列元素为E类型
  private val eventQueue: BlockingQueue[E] = new LinkedBlockingDeque[E]()

  // 标志位
  private val stopped = new AtomicBoolean(false)

  // 事件处理线程
  private val eventThread = new Thread(name) {
    // 设置为后台线程
    setDaemon(true)

    override def run(): Unit = {
      try {
        // 如果标志位stopped没有被设置为true,一直循环
        while (!stopped.get) {
          // 从事件队列中take一条事件
          val event = eventQueue.take()
          try {
            // 调用onReceive()方法进行处理
            onReceive(event)
          } catch {
            case NonFatal(e) => {
              try {
                onError(e)
              } catch {
                case NonFatal(e) => logError("Unexpected error in " + name, e)
              }
            }
          }
        }
      } catch {
        case ie: InterruptedException => // exit even if eventQueue is not empty
        case NonFatal(e) => logError("Unexpected error in " + name, e)
      }
    }

  }

  def start(): Unit = {
    if (stopped.get) {
      throw new IllegalStateException(name + " has already been stopped")
    }
    // Call onStart before starting the event thread to make sure it happens before onReceive
    onStart()
    eventThread.start()
  }

  def stop(): Unit = {
    if (stopped.compareAndSet(false, true)) {
      eventThread.interrupt()
      var onStopCalled = false
      try {
        eventThread.join()
        // Call onStop after the event thread exits to make sure onReceive happens before onStop
        onStopCalled = true
        onStop()
      } catch {
        case ie: InterruptedException =>
          Thread.currentThread().interrupt()
          if (!onStopCalled) {
            // ie is thrown from `eventThread.join()`. Otherwise, we should not call `onStop` since
            // it's already called.
            onStop()
          }
      }
    } else {
      // Keep quiet to allow calling `stop` multiple times.
    }
  }

  /**
   * Put the event into the event queue. The event thread will process it later.
   * 将事件加入到时间队列。事件线程过会会处理它。
   */
  def post(event: E): Unit = {
    // 将事件加入到待处理队列
    eventQueue.put(event)
  }

  /**
   * Return if the event thread has already been started but not yet stopped.
   */
  def isActive: Boolean = eventThread.isAlive

  /**
   * Invoked when `start()` is called but before the event thread starts.
   */
  protected def onStart(): Unit = {}

  /**
   * Invoked when `stop()` is called and the event thread exits.
   */
  protected def onStop(): Unit = {}

  /**
   * Invoked in the event thread when polling events from the event queue.
   *
   * Note: Should avoid calling blocking actions in `onReceive`, or the event thread will be blocked
   * and cannot process events in time. If you want to call some blocking actions, run them in
   * another thread.
   */
  protected def onReceive(event: E): Unit

  /**
   * Invoked if `onReceive` throws any non fatal error. Any non fatal error thrown from `onError`
   * will be ignored.
   */
  protected def onError(e: Throwable): Unit

}
        我们可以看到,EventLoop实际上就是一个任务队列及其对该队列一系列操作的封装。在它内部,首先定义了一个LinkedBlockingDeque类型的事件队列,队列元素为E类型,其中DAGSchedulerEventProcessLoop存储的则是DAGSchedulerEvent类型的事件,代码如下:
// LinkedBlockingDeque类型的事件队列,队列元素为E类型
  private val eventQueue: BlockingQueue[E] = new LinkedBlockingDeque[E]()
        并提供了一个后台线程,专门对事件队列里的事件进行监控,并调用onReceive()方法进行处理,代码如下:

// 事件处理线程
  private val eventThread = new Thread(name) {
    // 设置为后台线程
    setDaemon(true)

    override def run(): Unit = {
      try {
        // 如果标志位stopped没有被设置为true,一直循环
        while (!stopped.get) {
          // 从事件队列中take一条事件
          val event = eventQueue.take()
          try {
            // 调用onReceive()方法进行处理
            onReceive(event)
          } catch {
            case NonFatal(e) => {
              try {
                onError(e)
              } catch {
                case NonFatal(e) => logError("Unexpected error in " + name, e)
              }
            }
          }
        }
      } catch {
        case ie: InterruptedException => // exit even if eventQueue is not empty
        case NonFatal(e) => logError("Unexpected error in " + name, e)
      }
    }

  }
        那么如何向队列中添加事件呢?调用其post()方法,传入事件即可。如下:

/**
   * Put the event into the event queue. The event thread will process it later.
   * 将事件加入到时间队列。事件线程过会会处理它。
   */
  def post(event: E): Unit = {
    // 将事件加入到待处理队列
    eventQueue.put(event)
  }
        言归正传,上面提到,submitJob()方法利用eventProcessLoop的post()方法加入一个JobSubmitted事件到事件队列中,那么DAGSchedulerEventProcessLoop对于JobSubmitted事件是如何处理的呢?我们看它的onReceive()方法,源码如下:

/**
   * The main event loop of the DAG scheduler.
   * DAGScheduler中事件主循环
   */
  override def onReceive(event: DAGSchedulerEvent): Unit = {
    val timerContext = timer.time()
    try {
      // 调用doOnReceive()方法,将DAGSchedulerEvent类型的event传递进去
      doOnReceive(event)
    } finally {
      timerContext.stop()
    }
  }
        继续看doOnReceive()方法,代码如下:
// 事件处理调度函数
  private def doOnReceive(event: DAGSchedulerEvent): Unit = event match {
    
    // 如果是JobSubmitted事件,调用dagScheduler.handleJobSubmitted()方法处理
    case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties) =>
      dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)

    // 如果是MapStageSubmitted事件,调用dagScheduler.handleMapStageSubmitted()方法处理
    case MapStageSubmitted(jobId, dependency, callSite, listener, properties) =>
      dagScheduler.handleMapStageSubmitted(jobId, dependency, callSite, listener, properties)

    case StageCancelled(stageId) =>
      dagScheduler.handleStageCancellation(stageId)

    case JobCancelled(jobId) =>
      dagScheduler.handleJobCancellation(jobId)

    case JobGroupCancelled(groupId) =>
      dagScheduler.handleJobGroupCancelled(groupId)

    case AllJobsCancelled =>
      dagScheduler.doCancelAllJobs()

    case ExecutorAdded(execId, host) =>
      dagScheduler.handleExecutorAdded(execId, host)

    case ExecutorLost(execId) =>
      dagScheduler.handleExecutorLost(execId, fetchFailed = false)

    case BeginEvent(task, taskInfo) =>
      dagScheduler.handleBeginEvent(task, taskInfo)

    case GettingResultEvent(taskInfo) =>
      dagScheduler.handleGetTaskResult(taskInfo)

    case completion @ CompletionEvent(task, reason, _, _, taskInfo, taskMetrics) =>
      dagScheduler.handleTaskCompletion(completion)

    case TaskSetFailed(taskSet, reason, exception) =>
      dagScheduler.handleTaskSetFailed(taskSet, reason, exception)

    case ResubmitFailedStages =>
      dagScheduler.resubmitFailedStages()
  }
        对于JobSubmitted事件,我们通过调用DAGScheduler的handleJobSubmitted()方法来处理。

        好了,到这里,第一阶段Job的调度模型与运行反馈大体已经分析完了,至于后面的第二、第三阶段,留待后续博文继续分析吧~





相关文章
|
1月前
|
存储 分布式计算 监控
Spark作业的调度与执行流程
Spark作业的调度与执行流程
|
2月前
|
分布式计算 监控 Spark
Spark 任务运行时日志分析
Spark 任务运行时日志分析
23 0
|
2月前
|
设计模式 SQL 分布式计算
Spark Day06:Spark Core之Spark 内核调度和SparkSQL快速入门
Spark Day06:Spark Core之Spark 内核调度和SparkSQL快速入门
37 0
|
2月前
|
存储 分布式计算 大数据
【大数据技术Hadoop+Spark】Spark RDD设计、运行原理、运行流程、容错机制讲解(图文解释)
【大数据技术Hadoop+Spark】Spark RDD设计、运行原理、运行流程、容错机制讲解(图文解释)
44 0
|
2月前
|
存储 分布式计算 负载均衡
【大数据技术Hadoop+Spark】MapReduce概要、思想、编程模型组件、工作原理详解(超详细)
【大数据技术Hadoop+Spark】MapReduce概要、思想、编程模型组件、工作原理详解(超详细)
36 0
|
4月前
|
SQL 分布式计算 算法
【大数据处理框架】Spark大数据处理框架,包括其底层原理、架构、编程模型、生态圈
【大数据处理框架】Spark大数据处理框架,包括其底层原理、架构、编程模型、生态圈
114 0
|
5月前
|
分布式计算 搜索推荐 算法
大数据Spark MLlib基于模型的协同过滤
大数据Spark MLlib基于模型的协同过滤
47 0
|
8月前
|
存储 分布式计算 Spark
SPARK中关于HighlyCompressedMapStatus的说明(会造成运行时的数据不精确)
SPARK中关于HighlyCompressedMapStatus的说明(会造成运行时的数据不精确)
54 0
|
8月前
|
机器学习/深度学习 缓存 分布式计算
一种基于Spark深度随机森林的网络入侵检测模型
一种基于Spark深度随机森林的网络入侵检测模型
|
9月前
|
机器学习/深度学习 人工智能 分布式计算
SparK项目原作解读:卷积模型的首个BERT预训练
SparK项目原作解读:卷积模型的首个BERT预训练
156 0