Hadoop2源码分析-YARN 的服务库和事件库

1.概述

  在《Hadoop2源码分析-YARN RPC 示例介绍》一文当中,给大家介绍了YARN 的 RPC 机制,以及相关代码的演示,今天我们继续去学习 YARN 的服务库和事件库,分享目录如下所示:

  • 服务库和事件库介绍
  • 使用示例
  • 截图预览

  下面开始今天的内容分享。

2.服务库和事件库介绍

2.1服务库

  YARN对于生命周期较长的对象使用服务的对象模型进行管理,主要特点如下:

  • 用于被服务化的对象包含4个状态,他们分别是:被创建、已初始化、已启动和已停止。源代码地址在 org.apache.hadoop.service 的 Service 接口中,内容如下所示:
Hadoop2源码分析-YARN 的服务库和事件库
public enum STATE {
    /** Constructed but not initialized */
    NOTINITED(0, "NOTINITED"),

    /** Initialized but not started or stopped */
    INITED(1, "INITED"),

    /** started and not stopped */
    STARTED(2, "STARTED"),

    /** stopped. No further state transitions are permitted */
    STOPPED(3, "STOPPED");

    /**
     * An integer value for use in array lookup and JMX interfaces.
     * Although {@link Enum#ordinal()} could do this, explicitly
     * identify the numbers gives more stability guarantees over time.
     */
    private final int value;

    /**
     * A name of the state that can be used in messages
     */
    private final String statename;

    private STATE(int value, String name) {
      this.value = value;
      this.statename = name;
    }

    /**
     * Get the integer value of a state
     * @return the numeric value of the state
     */
    public int getValue() {
      return value;
    }

    /**
     * Get the name of a state
     * @return the state's name
     */
    @Override
    public String toString() {
      return statename;
    }
  }
Hadoop2源码分析-YARN 的服务库和事件库
public abstract class AbstractService implements Service {
  
  // ......

}

  通过阅读代码,我们可以看出,服务的对象它实现了接口Service,并定义了最基本的服务状态:创建、初始化、启动以及停止。对于 AbstractService 类来说,它实现了 Service 接口。

  • 任何服务状态的变化都可以触发其他的动作,例如:
Hadoop2源码分析-YARN 的服务库和事件库
public void start() {
    if (isInState(STATE.STARTED)) {
      return;
    }
    //enter the started state
    synchronized (stateChangeLock) {
      if (stateModel.enterState(STATE.STARTED) != STATE.STARTED) {
        try {
          startTime = System.currentTimeMillis();
          serviceStart();
          if (isInState(STATE.STARTED)) {
            //if the service started (and isn't now in a later state), notify
            if (LOG.isDebugEnabled()) {
              LOG.debug("Service " + getName() + " is started");
            }
            notifyListeners();
          }
        } catch (Exception e) {
          noteFailure(e);
          ServiceOperations.stopQuietly(LOG, this);
          throw ServiceStateException.convert(e);
        }
      }
    }
  }
Hadoop2源码分析-YARN 的服务库和事件库

  这里,我们会去触发一个监听动作,全局监听状态的改变,异常的捕捉监听等。

  • 可以通过组合的方式进行服务组合,这样做的好处是便于统一去管理:在 YARN 中,如果是非组合服务,可以直接继承 AbstractService 类,否则需继承 CompositeService。

2.2事件库

  在 YARN 中,核心服务其本质就是一个*异步调度器,包含有ResourceManager、 NodeManager、MRAppMaster等内容,YARN 事件与事件处理器的关系在 

org.apache.hadoop.yarn.event  中。在使用 YARN 事件库的时候,需要先定义一个*异步调度器 AsyncDispatcher,它负责事件的处理与转发,然后我们根据实际业务需求定义一系列事件 Event 与事件处理器 EventHandler,并将事件注册到*异步调度器中用于完成事件统一管理和应用调度。流程如下图所示:

Hadoop2源码分析-YARN 的服务库和事件库

3.使用示例

  接下来,我们编写示例代码,去代码中理解这部分流程。

  • 首先是 JMRAppMaster 类:
Hadoop2源码分析-YARN 的服务库和事件库
package cn.hadoop.task.exec;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.EventHandler;

import cn.hadoop.task.CompositeService;
import cn.hadoop.task.JobEvent;
import cn.hadoop.task.JobEventType;
import cn.hadoop.task.TaskEvent;
import cn.hadoop.task.TaskEventType;

/**
 * @Date Jul 22, 2015
 *
 * @Author dengjie
 *
 * @Note TODO
 */
public class JMRAppMaster extends CompositeService {
    private Dispatcher dispatcher; // AsyncDispatcher
    private String jobID;
    private int taskNumber; // include numbers
    private String[] taskIDs; // include all task

    public JMRAppMaster(String name, String jobID, int taskNumber) {
        super(name);
        this.jobID = jobID;
        this.taskNumber = taskNumber;
        taskIDs = new String[taskNumber];
        for (int i = 0; i < taskNumber; i++) {
            taskIDs[i] = new String(this.jobID + "_task_" + i);
        }
    }

    public void serviceInit(Configuration conf) throws Exception {
        dispatcher = new AsyncDispatcher();// default a AsyncDispatcher
        dispatcher.register(JobEventType.class, new JobEventDispatcher());// register a job
        dispatcher.register(TaskEventType.class, new TaskEventDispatcher());// register a task
        addService((Service) dispatcher);
        super.serviceInit(conf);
    }

    public Dispatcher getDispatcher() {
        return dispatcher;
    }

    private class JobEventDispatcher implements EventHandler<JobEvent> {

        @SuppressWarnings("unchecked")
        public void handle(JobEvent event) {
            if (event.getType() == JobEventType.JOB_KILL) {
                System.out.println("Receive JOB_KILL event, killing all the tasks");
                for (int i = 0; i < taskNumber; i++) {
                    dispatcher.getEventHandler().handle(new TaskEvent(taskIDs[i], TaskEventType.T_KILL));
                }
            } else if (event.getType() == JobEventType.JOB_INIT) {
                System.out.println("Receive JOB_INIT event, scheduling tasks");
                for (int i = 0; i < taskNumber; i++) {
                    dispatcher.getEventHandler().handle(new TaskEvent(taskIDs[i], TaskEventType.T_SCHEDULE));
                }
            }
        }
    }

    private class TaskEventDispatcher implements EventHandler<TaskEvent> {

        public void handle(TaskEvent event) {
            if (event.getType() == TaskEventType.T_KILL) {
                System.out.println("Receive T_KILL event of task id " + event.getTaskID());
            } else if (event.getType() == TaskEventType.T_SCHEDULE) {
                System.out.println("Receive T_SCHEDULE event of task id " + event.getTaskID());
            }
        }
    }
}
Hadoop2源码分析-YARN 的服务库和事件库

  另外,还需要添加一些其他类,这些类以来可以在 Hadoop 源码工程中找到,这里就不贴代码了,大家可以到 Hadoop 工程的源码中找到对应的类,相关类名如下图所示:

Hadoop2源码分析-YARN 的服务库和事件库

  接下来是一个测试类,去测试一下我们所编写的示例流程。

  • JMRAppMasterTest类:

Hadoop2源码分析-YARN 的服务库和事件库
package cn.hadoop.rpc.test.yarn.task;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.conf.YarnConfiguration;

import cn.hadoop.task.JobEvent;
import cn.hadoop.task.JobEventType;
import cn.hadoop.task.exec.JMRAppMaster;

/**
 * @Date Jul 22, 2015
 *
 * @Author dengjie
 *
 * @Note TODO
 */
public class JMRAppMasterTest {
    @SuppressWarnings({ "unchecked", "resource" })
    public static void main(String[] args) {
        String jobID = "job_20150723_11";
        JMRAppMaster appMaster = new JMRAppMaster("Simple MRAppMaster Test", jobID, 10);
        YarnConfiguration conf = new YarnConfiguration(new Configuration());
        try {
            appMaster.serviceInit(conf);
            appMaster.serviceStart();
        } catch (Exception e) {
            e.printStackTrace();
        }
        appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID, JobEventType.JOB_KILL));
        appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID, JobEventType.JOB_INIT));
    }
}
Hadoop2源码分析-YARN 的服务库和事件库

4.截图预览

  在编写完成相关流程代码后,我们运行代码来观察整个流程,截图如下所示:

Hadoop2源码分析-YARN 的服务库和事件库

5.总结

  在编写这部分流程代码时,可以参考 Hadoop YARN 部分的工程源码,通过运行调试代码,掌握对事件库和服务库的流程,以及它们的工作机制。另外,在编写的过程当中,最好将源码的文件引入到自己的工程,不要单独使用 JAR 包的方式导入,由于我们是独立运行某个模块,需要改动源代码的函数访问权限,若是直接引入 JAR 包地址,会导致函数修饰权限问题而不能运行,这里大家在运行调试的时候注意即可。

6.结束语

  这篇博客就和大家分享到这里,如果大家在研究学习的过程当中有什么问题,可以加群进行讨论或发送邮件给我,我会尽我所能为您解答,与君共勉!

联系方式: 
邮箱:smartloli.org@gmail.com 
Twitter:https://twitter.com/smartloli 
QQ群(Hadoop - 交流社区1):424769183 
温馨提示:请大家加群的时候写上加群理由(姓名+公司/学校),方便管理员审核,谢谢! 

热爱生活,享受编程,与君共勉!



本文转自哥不是小萝莉博客园博客,原文链接:http://www.cnblogs.com/smartloli/,如需转载请自行联系原作者

上一篇:怎样在 CentOS 里下载 RPM 包及其所有依赖


下一篇:【HDU 2586 How far away?】LCA问题 Tarjan算法