Hadoop2.6.0的事件分类与实现

简介: 说实在的,在阅读Hadoop YARN的源码之前,我对于java枚举的使用相形见绌。YARN中实现的事件在可读性、可维护性、可扩展性方面的工作都值得借鉴。

前言

说实在的,在阅读Hadoop YARN的源码之前,我对于java枚举的使用相形见绌。YARN中实现的事件在可读性、可维护性、可扩展性方面的工作都值得借鉴。

概念

在具体分析源码之前,我们先看看YARN是如何定义一个事件的。比如作业启动的事件,很多人可能会用常量将它定义到一个class文件中,就像下面这样:

class Constants {
  public static final String JOB_START_EVENT = "jobStart";
}

或者简单的使用枚举,就像下面这样;

enum Enums {
  JOB_START_EVENT("jobStart");
  private String name;
  private Enums(String name) {
    this.name = name;
  }
}

之后,当增加了作业停止的事件,代码会变为:

class Constants {
  public static final String JOB_START_EVENT = "jobStart";
  public static final String JOB_END_EVENT = "jobEnd";
}

或者:

enum Enums {
  JOB_START_EVENT("jobStart"),
  JOB_END_EVENT("jobEnd");
  private String name;
  private Enums(String name) {
    this.name = name;
  }
}

我们的系统往往很复杂,这时候引入了任务的概念,包括任务启动、任务停止的事件。随着业务发展,有更多的概念被加进来,就像下面这样;

class Constants {
  public static final String JOB_START_EVENT = "jobStart";
  public static final String JOB_END_EVENT = "jobEnd";
  public static final String TASK_START_EVENT = "taskStart";
  public static final String TASK_END_EVENT = "taskEnd";
  // 其它各种概念的常量
}

或者:

enum Enums {
  JOB_START_EVENT("jobStart"),
  JOB_END_EVENT("jobEnd"),
  // 其它各种概念的常量枚举
  TASK_START_EVENT("taskStart"),
  TASK_END_EVENT("taskEnd");
  private String name;
  private Enums(String name) {
    this.name = name;
  }
}

当加入的常量值越来越多时,你会发现以上使用方式越来越不可维护。各种概念混杂在一起,显得杂乱无章。你可能会说,我不会这么傻,我会将作业与任务以及其它概念的常量值分而治之,每个业务概念相关的放入一个文件,就像下面这样:

class JobConstants {
  public static final String JOB_START_EVENT = "jobStart";
  public static final String JOB_END_EVENT = "jobEnd";
}

class TaskConstants {
  public static final String TASK_START_EVENT = "taskStart";
  public static final String TASK_END_EVENT = "taskEnd";
}

或者:

enum JobEnums {
  JOB_START_EVENT("jobStart"),
  JOB_END_EVENT("jobEnd");
  private String name;
  private JobEnums (String name) {
    this.name = name;
  }
}

enum TaskEnums {
  TASK_START_EVENT("taskStart"),
  TASK_END_EVENT("taskEnd");
  private String name;
  private TaskEnums (String name) {
    this.name = name;
  }
}

现在业务出现了新的变化,每种枚举值除了name属性之外,还增加了code属性。假如你之前选择了常量值来实现,此时不可避免的需要重构。如果你选择了枚举,说明你初步的选择是明智的,你可以这样来扩展:

enum JobEnums {
  JOB_START_EVENT(10, "jobStart"),
  JOB_END_EVENT(20, "jobEnd");
  private int code;
  private String name;
  private JobEnums (int code, String name) {
    this.code = code;  
    this.name = name;
  }
}

enum TaskEnums {
  TASK_START_EVENT(110, "taskStart"),
  TASK_END_EVENT(120, "taskEnd");
  private int code;
  private String name;
  private TaskEnums (int code, String name)   {
    this.code = code;
    this.name = name;
  }
}

可悲的是,你不得不在每一个枚举中都重复加入类似的代码。也许你认为这只不过是增加些许的工作量,你操作键盘的手法熟练而迷人,几次快速的复制操作就可以完成。噩梦远没有结束,新的需求给两个枚举类型融入了新的不同——JobEnums增加了description属性,而TaskEnums则增加了timestamp字段。此外,两者还必须都增加hashCode方法以用于散列。增加这些功能后,代码将变为:

enum JobEnums {
  JOB_START_EVENT(10, "jobStart", "job start description"),
  JOB_END_EVENT(20, "jobEnd", "job end description");
  private int code;
  private String name;
  private String description;
  private JobEnums (int code, String name, String description) {
    this.code = code;  
    this.name = name;
    this.description = description;
  }

  public int hashCode() {
    return this.name.hashCode() + this.description.hashCode();
  }
}

enum TaskEnums {
  TASK_START_EVENT(110, "taskStart", 1460977775087),
  TASK_END_EVENT(120, "taskEnd", 1460977775088);
  private int code;
  private String name;
  private long timestamp;
  private TaskEnums (int code, String name, long timestamp)   {
    this.code = code;
    this.name = name;
    this.timestamp = timestamp;
  }

  public int hashCode() {
    return this.name.hashCode();
  }
}

随着业务的发展,你会发现你需要维护的枚举类型差异越来越多。即便它们之间有所不同,可是却有很多内容是重复的。为了解决枚举与常量在可读性、可维护性、可复用性、可扩展性等方面的问题,Hadoop将事件进行了以下定义:
事件 = 事件名称 + 事件类型
比如作业启动事件 = 作业事件 + 作业事件类型

事件与事件类型

Hadoop2.6.0中的事件多种多样,最为常见的包括:ContainerEvent、ApplicationEvent、JobEvent、RMAppEvent、RMAppAttemptEvent、TaskEvent、TaskAttemptEvent等。为了解决枚举与常量在可读性、可维护性、可复用性、可扩展性等方面的问题,Hadoop对事件进行了以下抽象:

/**
 * Interface defining events api.
 *
 */
@Public
@Evolving
public interface Event<TYPE extends Enum<TYPE>> {

  TYPE getType();
  long getTimestamp();
  String toString();
}

以上接口说明了任何一个具体事件都是一个枚举类型,而且有一个事件类型属性(用泛型标记TYPE表示),一个时间戳及toString()方法。
所有事件都有一个基本实现AbstractEvent,其实现如下:

/**
 * Parent class of all the events. All events extend this class.
 */
@Public
@Evolving
public abstract class AbstractEvent<TYPE extends Enum<TYPE>> 
    implements Event<TYPE> {

  private final TYPE type;
  private final long timestamp;

  // use this if you DON'T care about the timestamp
  public AbstractEvent(TYPE type) {
    this.type = type;
    // We're not generating a real timestamp here.  It's too expensive.
    timestamp = -1L;
  }

  // use this if you care about the timestamp
  public AbstractEvent(TYPE type, long timestamp) {
    this.type = type;
    this.timestamp = timestamp;
  }

  @Override
  public long getTimestamp() {
    return timestamp;
  }

  @Override
  public TYPE getType() {
    return type;
  }

  @Override
  public String toString() {
    return "EventType: " + getType();
  }
}

以JobEvent表示作业事件,其实现如下:

/**
 * This class encapsulates job related events.
 *
 */
public class JobEvent extends AbstractEvent<JobEventType> {

  private JobId jobID;

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

  public JobId getJobId() {
    return jobID;
  }

}

TaskEvent表示任务事件,其实现如下:

/**
 * this class encapsulates task related events.
 *
 */
public class TaskEvent extends AbstractEvent<TaskEventType> {

  private TaskId taskID;

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

  public TaskId getTaskID() {
    return taskID;
  }
}

事件类型属性(用泛型标记TYPE表示)在任务事件中对应的是TaskEventType,其实现如下:

/**
 * Event types handled by Task.
 */
public enum TaskEventType {

  //Producer:Client, Job
  T_KILL,

  //Producer:Job
  T_SCHEDULE,
  T_RECOVER,

  //Producer:Speculator
  T_ADD_SPEC_ATTEMPT,

  //Producer:TaskAttempt
  T_ATTEMPT_LAUNCHED,
  T_ATTEMPT_COMMIT_PENDING,
  T_ATTEMPT_FAILED,
  T_ATTEMPT_SUCCEEDED,
  T_ATTEMPT_KILLED
}

JobEventType类似,不再赘述。
这种实现将枚举与各种事件之间的差异(表现在属性和方法的不同)解耦,极大地扩展了可读性、可维护性,并且保留了相同逻辑的代码复用。

后记:个人总结整理的《深入理解Spark:核心思想与源码分析》一书现在已经正式出版上市,目前京东、当当、天猫等网站均有销售,欢迎感兴趣的同学购买。
_

京东(现有满150减50活动)):http://item.jd.com/11846120.html
当当:http://product.dangdang.com/23838168.html

相关文章
|
存储 消息中间件 Java
Hadoop-No.15之Flume基于事件的数据收集和处理
Flume是一种分布式的可靠开源系统,用于流数据的高效收集,聚集和移动.Flume通常用于移动日志数据.但是也能移动大量事件数据.如社交媒体订阅,消息队列事件或者网络流量数据. Flume架构 Flume的数据源使用来自外部数据源的时间,然后转发到Channel中.
962 0
|
存储 分布式计算 Hadoop
Hadoop源码分类概要整理
  最近突然觉得, 很多掌握的都还是很浅的原理,需要更深入细粒度去了解整个分布式系统的运转机制。于是。。开始作死而又作死而又作死的源码之旅。   Hadoop包的功能总共有下列几类:   tool:提供一些命令行工具,如DistCp,archive   mapreduce,:Hadoop的Map/R...
1062 0
|
分布式计算 大数据 Hadoop
【Hadoop Summit Tokyo 2016】企业数据分类和治理
本讲义出自Shwetha Shivalingamurthy与Suma Shivaprasad在Hadoop Summit Tokyo 2016上的演讲,主要分享了企业数据分类和治理的案例并且深入地讲解了大数据治理的相关内容,并介绍了Atlas的概览和架构设计以及其特性和发展路线。
1678 0
|
资源调度 分布式计算 Java
Hadoop Yarn事件处理框架源码分析
由于想在项目中使用类似yarn的事件处理机制,就看了实现。主要是由Dispatcher.java,EventHandler.java,Service.java这3个类撑起来的。
1120 0
|
SQL 关系型数据库 Apache
2015也过去一半了,Hadoop大事件盘点
2015也快过去一半了,Hadoop在过去一年的发展究竟如何,下面小象带你盘点一下2014Hadoop大事件! 2014年2月,Hadoop 2.3.0发布,新特性包括支持HDFS的混合存储分级,可以集中管理HDFS内存里的缓存数据,通过HDFS中的YARN分布式缓存简化MapReduce分配及一些Bug修正。
1832 0
|
2月前
|
分布式计算 Kubernetes Hadoop
大数据-82 Spark 集群模式启动、集群架构、集群管理器 Spark的HelloWorld + Hadoop + HDFS
大数据-82 Spark 集群模式启动、集群架构、集群管理器 Spark的HelloWorld + Hadoop + HDFS
168 6
|
2月前
|
分布式计算 资源调度 Hadoop
大数据-80 Spark 简要概述 系统架构 部署模式 与Hadoop MapReduce对比
大数据-80 Spark 简要概述 系统架构 部署模式 与Hadoop MapReduce对比
69 2
|
28天前
|
存储 分布式计算 Hadoop
数据湖技术:Hadoop与Spark在大数据处理中的协同作用
【10月更文挑战第27天】在大数据时代,数据湖技术凭借其灵活性和成本效益成为企业存储和分析大规模异构数据的首选。Hadoop和Spark作为数据湖技术的核心组件,通过HDFS存储数据和Spark进行高效计算,实现了数据处理的优化。本文探讨了Hadoop与Spark的最佳实践,包括数据存储、处理、安全和可视化等方面,展示了它们在实际应用中的协同效应。
88 2

相关实验场景

更多