Yarn源码分析之MRAppMaster上MapReduce作业处理总流程(二)-阿里云开发者社区

开发者社区> 大数据> 正文
登录阅读全文

Yarn源码分析之MRAppMaster上MapReduce作业处理总流程(二)

简介:         本文继《Yarn源码分析之MRAppMaster上MapReduce作业处理总流程(一)》,接着讲述MapReduce作业在MRAppMaster上处理总流程,继上篇讲到作业初始化之后的作业启动,关于作业初始化主体流程的详细介绍,请参见《Yarn源码分析之MRAppMaster上MapReduce作业初始化解析》一文。

        本文继《Yarn源码分析之MRAppMaster上MapReduce作业处理总流程(一)》,接着讲述MapReduce作业在MRAppMaster上处理总流程,继上篇讲到作业初始化之后的作业启动,关于作业初始化主体流程的详细介绍,请参见《Yarn源码分析之MRAppMaster上MapReduce作业初始化解析》一文。

        (三)启动

        作业的启动是通过MRAppMaster的startJobs()方法实现的,其代码如下:

  /**
   * This can be overridden to instantiate multiple jobs and create a 
   * workflow.
   *
   * TODO:  Rework the design to actually support this.  Currently much of the
   * job stuff has been moved to init() above to support uberization (MR-1220).
   * In a typical workflow, one presumably would want to uberize only a subset
   * of the jobs (the "small" ones), which is awkward with the current design.
   */
  @SuppressWarnings("unchecked")
  protected void startJobs() {
    /** create a job-start event to get this ball rolling */
	// 构造作业启动事件JobStartEvent实例startJobEvent
    JobEvent startJobEvent = new JobStartEvent(job.getID(),
        recoveredJobStartTime);
    /** send the job-start event. this triggers the job execution. */
    // 将作业启动事件JobStartEvent实例startJobEvent交由事件分发器dispatcher的事件处理器处理
    dispatcher.getEventHandler().handle(startJobEvent);
  }
        很简单,首先构造作业启动事件JobStartEvent实例startJobEvent,然后将作业启动事件JobStartEvent实例startJobEvent交由事件分发器dispatcher的事件处理器处理。我们首先看下事件分发器dispatcher是如何初始化的,其在MRAppMaster服务初始化的serviceInit()方法中,关键代码如下:

dispatcher = createDispatcher();
        再来看下createDispatcher()方法,如下:

  protected Dispatcher createDispatcher() {
    return new AsyncDispatcher();
  }
        就是创建一个AsyncDispatcher对象,其代表的是一个事件异步分发器AsyncDispatcher,我们曾经在《Yarn源码分析之事件异步分发器AsyncDispatcher》一文中专门介绍过这个AsyncDispatcher。AsyncDispatcher其实是一个生产者-消费者模型的事件异步分发器。在其内部有一个待分发事件队列eventQueue,并有一个GenericEventHandler类型的事件处理器handlerInstance,由其handle()方法负责将外部事件event添加到待分发队列eventQueue中,等到AsyncDispatcher中的消费者eventHandlingThread不断的获取待分发队列eventQueue中的事件,分发并交由之前注册的事件类型对应的事件处理器处理。关于这部分的内容请阅读《Yarn源码分析之事件异步分发器AsyncDispatcher》一文,此处不再做过多介绍。

        那么dispatcher中是如何注册JobStartEvent事件的处理器的呢?注册的事件处理器又是谁呢?还是在服务初始化的方法中,如下:

      //register the event dispatchers
      dispatcher.register(JobEventType.class, jobEventDispatcher);
        通过查看JobStartEvent的源码我们知道,JobStartEvent继承自JobEvent,它也是一种JobEvent,所以其处理会交给jobEventDispatcher来处理。细心的读者获取会发现,在此之前,dispatcher已经注册过一个JobEventType对应的事件处理器,NoopEventHandler类型的eater了,代码如下:

      NoopEventHandler eater = new NoopEventHandler();
      //We do not have a JobEventDispatcher in this path
      dispatcher.register(JobEventType.class, eater);
        我们先看下NoopEventHandler的定义,如下:

  /**
   * Eats events that are not needed in some error cases.
   */
  private static class NoopEventHandler implements EventHandler<Event> {

    @Override
    public void handle(Event event) {
      //Empty
    }
  }
        四个字,空空如也!那么,读者在这里可能就有疑问了,到底是由jobEventDispatcher还是eater来处理作业启动JobStartEvent事件内。这里要说的是,这两次注册实际上是形成了一个JobEventType事件类型的链式事件处理器,它会将事件挨个通过链式事件处理器中的每个处理器进行处理,这在《Yarn源码分析之事件异步分发器AsyncDispatcher》一文中的register()方法介绍中也提到过了,读者可自行查看。而这里,既然eater为空,不对事件做任何处理,我们还是看看jobEventDispatcher吧。

        那么,jobEventDispatcher是如何定义及初始化的呢?其实这个jobEventDispatcher在Yarn源码分析之MRAppMaster上MapReduce作业处理总流程(一)中的作业初始化事件JobEventType.JOB_INIT处理时已经讲到过了,它就是一个JobEventDispatcher对象,这里再回顾一下,其定义如下:

  private class JobEventDispatcher implements EventHandler<JobEvent> {
    @SuppressWarnings("unchecked")
    @Override
    public void handle(JobEvent event) {
      // 从应用运行上下文信息context中根据jobId获取Job实例,即JobImpl对象,调用其handle()方法,处理对应事件
      ((EventHandler<JobEvent>)context.getJob(event.getJobId())).handle(event);
    }
  }
        它实际上并不真正干活,而是从应用运行上下文信息context中根据jobId获取Job实例,即JobImpl对象,调用其handle()方法,处理对应事件。那么在处理作业初始化事件时我们也提到过了,它是根据作业状态机的doTransition()方法根据事件类型来处理的,关于作业状态机,我们这里还是不做介绍,你还是只要知道作业启动事件是通过JobImpl的静态内部类StartTransition的transition()方法来处理的就行,其代码如下:
  public static class StartTransition
  implements SingleArcTransition<JobImpl, JobEvent> {
    /**
     * This transition executes in the event-dispatcher thread, though it's
     * triggered in MRAppMaster's startJobs() method.
     */
    @Override
    public void transition(JobImpl job, JobEvent event) {
      JobStartEvent jse = (JobStartEvent) event;
     
      // 设置作业的起始时间startTime
      if (jse.getRecoveredJobStartTime() != 0) {
        job.startTime = jse.getRecoveredJobStartTime();
      } else {
        job.startTime = job.clock.getTime();
      }
      
      // 创建作业已初始化事件JobInitedEvent实例jie
      JobInitedEvent jie =
        new JobInitedEvent(job.oldJobId,
             job.startTime,
             job.numMapTasks, job.numReduceTasks,
             job.getState().toString(),
             job.isUber());
      
      // 将作业已初始化事件JobInitedEvent实例jie包装成作业历史事件JobHistoryEvent,并交给作业的事件处理器eventHandler处理
      job.eventHandler.handle(new JobHistoryEvent(job.jobId, jie));
      
      // 创建作业信息变更事件JobInfoChangeEvent实例jice
      JobInfoChangeEvent jice = new JobInfoChangeEvent(job.oldJobId,
          job.appSubmitTime, job.startTime);
      
      // 将作业信息变更事件JobInfoChangeEvent实例jice包装成作业历史事件JobHistoryEvent,并交给作业的事件处理器eventHandler处理
      job.eventHandler.handle(new JobHistoryEvent(job.jobId, jice));
      
      // 调用作业度量指标metrics的runningJob()方法,标识作业已开始运行
      job.metrics.runningJob(job);

      // 构造提交作业Setup事件CommitterJobSetupEvent,并交由作业的事件处理器eventHandler处理
      job.eventHandler.handle(new CommitterJobSetupEvent(
              job.jobId, job.jobContext));
    }
  }
        去掉关于作业历史信息等不是十分关键的细节,整体主体流程如下:

        1、设置作业的起始时间startTime;

        2、构造提交作业Setup事件CommitterJobSetupEvent,并交由作业的事件处理器eventHandler处理。

        那么,作业的事件处理器eventHandler是什么呢?它又是如何处理提交作业Setup事件CommitterJobSetupEvent的呢?

        我们先看下作业的事件处理器eventHandler,在MRAppMaster服务启动时创建作业JobImpl实例时,eventHandler是通过传入的dispatcher.getEventHandler()来初始化的,基于上面的陈述,这我们就不用讲了吧。

        我们还是看下dispatcher是如何注册事件CommitterJobSetupEvent对应的事件处理器的吧,代码如下:

     dispatcher.register(CommitterEventType.class, committerEventHandler);
        我们知道,CommitterJobSetupEvent继承自CommitterEvent,所以它实际上是通过committerEventHandler来处理的,那么什么是committerEventHandler呢?其初始化如下:

      //service to handle the output committer
      committerEventHandler = createCommitterEventHandler(context, committer);
        通过调用createCommitterEventHandler()方法,构造了一个CommitterEventHandler实例,如下:

  protected EventHandler<CommitterEvent> createCommitterEventHandler(
      AppContext context, OutputCommitter committer) {
    return new CommitterEventHandler(context, committer,
        getRMHeartbeatHandler(), jobClassLoader);
  }
        关于CommitterEventHandler的介绍,我们后续会写相关文章进行详细的介绍,这里,你只要知道,它类似AsyncDispatcher,也是一个生产者-消费者模式的事件分发器,而最终是通过其内部EventProcessor类型的事件处理线程eventHandlingThread来处理的,在EventProcessor中,有针对JOB_SETUP事件处理的逻辑,关键代码如下:

      switch (event.getType()) {
      case JOB_SETUP:
        handleJobSetup((CommitterJobSetupEvent) event);
        break;
        继续追踪handleJobSetup()方法,如下:

    @SuppressWarnings("unchecked")
    protected void handleJobSetup(CommitterJobSetupEvent event) {
      try {
        committer.setupJob(event.getJobContext());
        context.getEventHandler().handle(
            new JobSetupCompletedEvent(event.getJobID()));
      } catch (Exception e) {
        LOG.warn("Job setup failed", e);
        context.getEventHandler().handle(new JobSetupFailedEvent(
            event.getJobID(), StringUtils.stringifyException(e)));
      }
    }
        它做了两件事情,如下:

        1、调用committer的setupJob()方法处理该CommitterJobSetupEvent事件;

        2、又构造了一个JobSetupCompletedEvent事件,交由应用运行上下文context的事件处理器进行处理。

        而首先要说的是,committer、context均是由MRAppMaster在创建CommitterEventHandler时传入的,其对应的对象类型分别是:

        1、committer:

              1.1、新版API是通过OutputFormat组件的getOutputCommitter()方法获取的;

              1.2、旧版API是通过参数mapred.output.committer.class获取的,参数未配置默认为FileOutputCommitter。

        2、context:RunningAppContext。

        对于committer,我们这里以较为通用的FileOutputCommitter为例,看下其setupJob()方法,如下:

  /**
   * Create the temporary directory that is the root of all of the task 
   * work directories.
   * @param context the job's context
   */
  public void setupJob(JobContext context) throws IOException {
    if (hasOutputPath()) {
      Path jobAttemptPath = getJobAttemptPath(context);
      FileSystem fs = jobAttemptPath.getFileSystem(
          context.getConfiguration());
      if (!fs.mkdirs(jobAttemptPath)) {
        LOG.error("Mkdirs failed to create " + jobAttemptPath);
      }
    } else {
      LOG.warn("Output Path is null in setupJob()");
    }
  }
        实际上就做了一件事情,创建作业中所有任务工作的临时根目录。

        再来看下context是如何处理JobSetupCompletedEvent的,还记得之前我们讲述的,RunningAppContext实际上什么都不干,而是交给了JobImpl对应的作业状态机了吗?我们就看下JobImpl中是如何处理JobSetupCompletedEvent事件的,其对应的处理在其静态内部类SetupCompletedTransition的transition()中,代码如下:

  private static class SetupCompletedTransition
      implements SingleArcTransition<JobImpl, JobEvent> {
    @Override
    public void transition(JobImpl job, JobEvent event) {
      
      // 通过设置作业的setupProgress为1,标记作业setup已完成
      job.setupProgress = 1.0f;
      
      // 调度作业的Map Task
      job.scheduleTasks(job.mapTasks, job.numReduceTasks == 0);
      // 调度作业的Reduce Task
      job.scheduleTasks(job.reduceTasks, true);

      // If we have no tasks, just transition to job completed
      // 如果没有task了,则生成JOB_COMPLETED事件并交由作业的事件处理器eventHandler进行处理
      if (job.numReduceTasks == 0 && job.numMapTasks == 0) {
        job.eventHandler.handle(new JobEvent(job.jobId,
            JobEventType.JOB_COMPLETED));
      }
    }
  }
        是不是很简单,而且也很理所当然?处理流程如下:

        1、通过设置作业的setupProgress为1,标记作业setup已完成;

        2、调度作业的Map Task;

        3、调度作业的Reduce Task;

        4、如果没有task了,则生成JOB_COMPLETED事件并交由作业的事件处理器eventHandler进行处理。


        未完待续,后续作业启动部分内容详细描述、作业停止等内容,请关注《Yarn源码分析之MRAppMaster上MapReduce作业处理总流程(三)》。

 













版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
大数据
使用钉钉扫一扫加入圈子
+ 订阅

大数据计算实践乐园,近距离学习前沿技术

其他文章