关于hadoop:深入浅出-Yarn-架构与实现23-Yarn-基础库-服务库与事件库

2次阅读

共计 6893 个字符，预计需要花费 18 分钟才能阅读完成。

一个宏大的分布式系统，各个组件间是如何协调工作的？组件是如何解耦的？线程运行如何更高效，缩小阻塞带来的低效问题？本节将对 Yarn 的服务库和事件库进行介绍，看看 Yarn 是如何解决这些问题的。

一、服务库

一）简介

对于生命周期较长的对象，Yarn 采纳基于服务的模型对其进行治理，有以下几个特点：

  • 基于状态治理：分为 4 个状态：NOTINITED(被创立)、INITED(已初始化)、STARTED(已启动)、STOPPED(已进行)。
  • 服务状态的变动会触发其余的操作。
  • 可通过组合的形式对服务进行组合。

二）源码简析

源代码地址在 hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/service 的 Service 接口中。
其中定义了服务的四个状态，以及须要实现的状态转换、获取信息、注册等办法。

public interface Service extends Closeable {

  public enum STATE {NOTINITED(0, "NOTINITED"),
    INITED(1, "INITED"),
    STARTED(2, "STARTED"),
    STOPPED(3, "STOPPED");
  }

  void init(Configuration config);
  void start();
  void stop();
  void close() throws IOException;
  void registerServiceListener(ServiceStateChangeListener listener);
  // ......

抽象类 AbstractService 实现了 Service 接口，提供了根底的 Service 实现，非组合服务间接继承这个抽象类再开发即可。

public abstract class AbstractService implements Service {
  // 以 start 实现为例，执行后会触发其余的操作
  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);
        }
      }
    }
  }

  // ......

对于组合类的服务如 ResourceManager、NodeManager 等，须要继承 CompositeService。其中会有对组合服务的逻辑解决。

  public List<Service> getServices() {synchronized (serviceList) {return new ArrayList<Service>(serviceList);
    }
  }

  protected void addService(Service service) {if (LOG.isDebugEnabled()) {LOG.debug("Adding service" + service.getName());
    }
    synchronized (serviceList) {serviceList.add(service);
    }
  }

二、事件库

传统函数式调用的问题：
整个执行过程是串行、同步进行的。调用另一个函数的时候，须要期待函数执行结束，才会持续往下走。示意图如下：

为了解决函数式调用的问题，可应用 「事件驱动」 的编程模型。

  • 所有对象都被形象成事件处理器
  • 事件处理器之间通过事件相关联
  • 每种事件处理器解决一种事件
  • 依据须要会触发另一种事件
  • 每类事件的解决可宰割为多个步骤，用无限状态机示意
  • 重要的是有一个 「地方异步调度器（AsyncDispatcher）」， 负责看待处理事件的收取和散发

示意图如下：

通过以上的形式，能够使程序有低耦合高内聚的特点，各个模块仅需实现各自的性能，同时进步了执行效率，把拆分的操作通过事件的形式发送进来即可。

三、服务库和事件库应用案例

本节将实现一个简化版的 MapReduce ApplicationMaster，帮忙理解 service 和 event 的应用办法。
与 MR 相似，一个 job 将被分为多个 task 执行。因而波及 job 和 task 两种对象的事件。并有一个 AsyncDispatcher 解决调度。
案例已上传至 github，有帮忙能够点个 ⭐️
https://github.com/Simon-Ace/hadoop-yarn-study-demo/tree/master/service-event-demo

一）事件局部

参考 hadoop 源码中 Task 和 Job Event 的实现，进行一些简化。
1、task

public enum TaskEventType {

  //Producer:Client, Job
  T_KILL,

  //Producer:Job
  T_SCHEDULE
}
public class TaskEvent extends AbstractEvent<TaskEventType> {

  private String taskID;

  public TaskEvent(String taskID, TaskEventType type) {super(type);
    this.taskID = taskID;
  }

  public String getTaskID() {return taskID;}
}

2、job

public enum JobEventType {

  //Producer:Client
  JOB_KILL,

  //Producer:MRAppMaster
  JOB_INIT
}
public class JobEvent extends AbstractEvent<JobEventType> {

    private String jobID;

    public JobEvent(String jobID, JobEventType type) {super(type);
        this.jobID = jobID;
    }

    public String getJobId() {return jobID;}
}

二）事件调度器

  • 定义和注册 EventDispatcher

  • service 初始化和启动办法

    import com.shuofxz.event.JobEvent;
import com.shuofxz.event.JobEventType;
import com.shuofxz.event.TaskEvent;
import com.shuofxz.event.TaskEventType;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.service.CompositeService;
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;

@SuppressWarnings("unchecked")
public class MyMRAppMaster extends CompositeService {
  private Dispatcher dispatcher;  // AsyncDispatcher
  private String jobID;
  private int taskNumber;         // 一个 job 蕴含的 task 数
  private String[] taskIDs;

  public MyMRAppMaster(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] = this.jobID + "_task_" + i;
      }
  }

  public void serviceInit(Configuration conf) throws Exception {dispatcher = new 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 void serviceStart() throws Exception {super.serviceStart();
  }

  public Dispatcher getDispatcher() {return dispatcher;}

  private class JobEventDispatcher implements EventHandler<JobEvent> {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());
          }
      }
  }
}

三）测试程序

  • 生成一个新的 job

  • 触发事件 JOB_KILL 和 JOB_INIT

    public class MyMRAppMasterTest {public static void main(String[] args) {
      String jobID = "job_20221011_99";
      MyMRAppMaster appMaster = new MyMRAppMaster("My 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));
  }
}

    输入后果：

    Receive JOB_KILL event, killing all the tasks
Receive JOB_INIT event, scheduling tasks
Receive T_KILL event of task id job_20150723_11_task_0
Receive T_KILL event of task id job_20150723_11_task_1
Receive T_KILL event of task id job_20150723_11_task_2
Receive T_KILL event of task id job_20150723_11_task_3
Receive T_KILL event of task id job_20150723_11_task_4
Receive T_KILL event of task id job_20150723_11_task_5
Receive T_KILL event of task id job_20150723_11_task_6
Receive T_KILL event of task id job_20150723_11_task_7
Receive T_KILL event of task id job_20150723_11_task_8
Receive T_KILL event of task id job_20150723_11_task_9
Receive T_SCHEDULE event of task id job_20150723_11_task_0
Receive T_SCHEDULE event of task id job_20150723_11_task_1
Receive T_SCHEDULE event of task id job_20150723_11_task_2
Receive T_SCHEDULE event of task id job_20150723_11_task_3
Receive T_SCHEDULE event of task id job_20150723_11_task_4
Receive T_SCHEDULE event of task id job_20150723_11_task_5
Receive T_SCHEDULE event of task id job_20150723_11_task_6
Receive T_SCHEDULE event of task id job_20150723_11_task_7
Receive T_SCHEDULE event of task id job_20150723_11_task_8
Receive T_SCHEDULE event of task id job_20150723_11_task_9

四、总结

本节介绍了 Yarn 的服务和事件库。
服务库标准了生命周期较长的服务型对象，定义了服务的四种状态、启停注册等要实现的办法，给出了繁多类型和组合类型服务的根本实现。
事件库的应用，解决了原始函数型调用的高耦合、阻塞低效等问题。可将一个大工作拆分成多个小工作，小工作变成不同的事件来触发解决。每一个事件处理器解决一种事件，并有一个地方异步调度器治理事件的收集和散发。
最初用一个简化的 MR ApplicationMaster 将事件库和服务库进行联合，更深领会如何在我的项目中将其联合应用。
学习过程中，写一个 demo 能更好的帮忙你了解常识。

参考文章：
《Hadoop 技术底细 – 深刻解析 Yarn 结构设计与实现原理》3.4 节

正文完
hadoop
发表至： hadoop
2022-10-13
 0
关于hadoop:macbook搭建hadoop伪分布记录
关于hadoop:深入浅出-Yarn-架构与实现62-NodeManager-状态机管理
关于hadoop:Hadoop-入门笔记-十-二-HDFS-Federation联邦机制
Giraph源码分析七-添加消息统计功能

站内搜索

-「」