YARN底层基础库

 

yarn基础库是其他一切模块的基础,它的设计直接决定了yarn的稳定性和扩展性,yarn借用了mrv1的一些底层基础库,比如rpc库等,但因为引入了很多新的软件设计方式,所以它的基础库更多,包括直接使用了开源序列化框架protocol buffers和apache avro,自定义的服务库、事件库和状态机等

---

目录

 

一. 概述

二. protocol buffers

三. apache avro
四. 底层通信库
五. 服务库与事件库

六. 状态机库

---

一. 概述

 

yarn基础库是其他一切模块的基础,它的设计直接决定了yarn的稳定性和扩展性

 

yarn的基础库主要有 : 

• protocol buffers : protocol buffers是google开源的序列化库,具有平台无关,高性能,兼容好等优点.yarn将protocolbuffers用到rpc通信中,默认情况下,yarn rpc中所有参数采用protocol buffers进行序列化/反序列化
• apache avro : avro是hadoop生态系统中的rpc框架,具有平台无关,支持动态模式等优点,avro的最初设计动机是解决yarn rpc兼容性和扩展性差等问题
• rpc库 : yarn采用mr1中的rpc库,但其中采用的默认序列化方法被替换成了protocol buffers
• 服务库和事件库 : yarn将所有的对象服务化,以便统一管理(创建,销毁等),而服务之间则采用事件机制进行通信
• 状态机库 : 状态机是一种表示有限个状态以及在这些状态之间的转移和动作等行为的数学模型

二. protocol buffers

 

protocol buffers是一种轻便高效的结构化数据存储格式,可以用于结构化数据序列化/反序列化

 

他适合做数据存储或rpc的数据交换格式,常用作通信协议,数据存储等领域的与语言无关,平台无关,可扩展的序列化结构数据格式

 

相比xml格式,protocol buffers的优点 : 

• 平台无关,语言无关
• 高性能,解析速度是xml的20 ~ 100倍
• 体积小,文件大小仅是xml的1/10 ~ 1/3
• 使用简单
• 兼容性好

 

yarn中,所有的rpc函数的参数均采用protocol buffers定义的,yarn的rpc协议全是使用protocol buffers定义(rpc协议上一章有介绍)

 

三. apache avro

 

apache avro是hadoop下的一个子项目。它本身既是一个序列化框架，同时也实现了rpc的功能

 

avro的特性和功能 ： 

• 丰富的数据结构类型
• 快速可压缩的二进制数据形式
• 存储持久数据的文件容器
• 提供远程过程调用rpc
• 简单的动态语言结合功能

 

avro具有以下特点 ： 

• 支持动态模式。avro不需要生成代码，有利于搭建通用的数据处理系统，避免了代码入侵
• 数据无需加标签
• 无需手工分配的域标识

 

avro作为日志序列化库使用，在yarn mapreduce中，所有事件的序列化/反序列化均采用avro完成

 

四. 底层通信库

  

hadooprpc的解析参考我的文章hadoop rpc机制详解

 

yarn提供的对外类是yarn rpc，用户只需使用该类便可以构建一个基于hadooprpc且采用protocol buffers序列化框架的通信协议

 

五. 服务库与事件库

 

服务库

 

对于生命周期较长的对象,yarn采用了基于服务的对象管理模型对其进行管理,该模型主要有以下几个特点 : 

• 将每个被服务化的对象分为4个状态 : notinited(被创建)，inited(已初始化)，started(已启动)，stopped(已停止)
• 任何服务状态变化都可以触发另外一些动作
• 可通过组合的方式对任意服务进行组合，以便进行统一管理

 

yarn中所有的服务对象最终都实现了接口service，它定义了最基本的服务初始化、启动、停止等操作，而abstractservice类提供了一个最基本的service实现。

 

yarn中所有对象，如果是非组合服务，直接继承abstractservice类即可，否则需compositeservice。比如，对于rm而言，它是一个组合服务，它组合了各种服务对象，包括clientrmservice、applicationmasterlauncher、applicationmasterservice等

 

 

　　　　　　　　　　　　　　　      yarn中服务模型的类图

 

yarn中，rm和nm属于组合服务，它们内部包含多个单一服务和组合服务，以实现对内部多种服务的统一管理

 

事件库

 

yarn采用了基于事件驱动的并发模型，该模型能够大大增强并发性，从而提高系统整体性能。为了构建该模型，yarn将各种处理逻辑抽象成事件和对应事件调度器，并将每类事件的处理过程分割成多个步骤，用有限状态机表示

 

 

　　　　　　　　　　　　　　　　　　　 yarn的事件处理模型

 

整个处理过程大致为 ：处理请求会作为事件进入系统，由*异步调度器(asyncdispatcher)负责传递给相应事件调度器(event handler)。该事件调度器可能将该事件转发给另外一个事件调度器，也可能交给一个带有有限状态机的事件处理器，其处理结果也以事件的形式输出给*异步调度器。而新的事件会再次被*异步调度器转发给下一个事件调度器，直至处理完成(达到终止条件)

 

yarn中，所有核心服务实际上都是一个*异步调度器，包括rm、nm和appmaster等，它们维护了事先注册的事件与事件处理器，并根据接收的事件类型驱动服务的运行

 

使用yarn事件库时，通常先定义一个*异步调度器，负责事件的转发与处理，然后根据实际业务需求定义一系列的事件与事件处理器，并注册到*异步调度器实现事件统一管理和调度。以mrappmaster为例，它内部包含一个*异步调度器，并注册了taskattemptevent/taskattemptimpl、taskevent/taskimpl、jobevent/jobimpl等一系列事件/事件处理器，由*异步调度器统一管理和调度

 

服务化和事件驱动软件设计思想的引入，使得yarn具有低耦合、高内聚的特点，各个模块只需完成各自功能，而模块之间则采用事件联系起来，系统设计简单且维护方便

 

yarn服务库和事件库的使用

 

yarn服务库和事件库的使用方法，介绍一个简单的实例，该例子涉及任务和作业两种对象的事件以及一个*异步调度器

 

(1) 定义task事件

 

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;
    }
}
// task事件类型定义
public enum taskeventtype {
    t_kill,
    t_schedule
}

 

(2) 定义job事件

 

public class jobevent extends abstractevent<jobeventtype> {
    private string jobid;
    public jobevent (string jobid, jobeventtype type) {
        super(type);
        this.jobid = jobid;
    }
}
//job事件类型定义
public enum jobeventtype {
    job_kill,
    job_init,
    job_start
}

 

(3) 事件调度器

 

定义一个*异步调度器，它接收job和task两种类型事件，并交给对应的事件处理器处理

@suppresswarnings("unchecked")
public class simplemrappmaster extends compositeservice {
    private dispatcher dispatcher; //*异步调度器
    private string jobid;
    private int tasknumber; //该作业包含的任务数目
    private string[] taskids; //该作业内部包含的所有任务
 
    public simplemrappmaster (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 (jobid + "_task_" + i);
        }
    }
 
    public void serviceinit (final configuration conf) throws exception {
        dispatcher = new asyncdispatcher(); //定义一个*异步调度器
        //分别注册job和task事件调度器
        dispatcher.register(jobeventtype.class, new jobeventdispatcher());
        dispatcher.register(taskeventtype.class, new taskeventdispatcher());
        addservice((service)dispatcher);
        super.serviceinit(conf);
    }
 
    public dispatcher getdispatcher() {
        return dispatcher;
    }
 
    private class jobeventdispatcher implements eventhandler<jobevent> {
        @override
        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> {
        @override
        public void handler (taskevent event) {
            if (event.gettype() == taskeventtype.t_kill) {
                system.out.println("receive t_kill event of task" + event.gettaskid());
            } else if (event.gettype() == taskeventtype.t_schedule) {
                system.out.println("receive t_schedule of task" + event.gettaskid()); 
            }
        }
    }
}

 

(4). 测试程序

 

@suppresswarnings("unchecked")
public class simplemrappmastertest {
    public static void main (string[] args) throws exception {
        string jobid = "job_20131215_12";
        simplemrappmaster appmaster = new simplemrappmaster("simple mrappmaster", jobid, 5);
        yarnconfiguration conf = new yarnconfiguration(new configuration());
        appmaster.serviceinit(conf);
        appmaster.servicestart();
        appmaster.getdispatcher().geteventhandler().handle(new jobevent(jobid, jobeventtype.job_kill));
        appmaster.getdispatcher().geteventhandler().handle(new jobevent(jobid, jobeventtype.job_init));
    }
}

 

事件驱动带来的变化

 

mrv1中，对象之间的作用关系是基于函数调用实现的，当一个对象向另外一个对象传递消息时，会直接采用函数调用的方式，且整个过程是串行的

 

基于函数调用的编程模型时低效的，它隐含着整个过程是串行、同步进行的。mrv2引入了事件驱动编程模型是一种更加高效的方式。

 

在基于事件驱动的编程模型中，所有对象被抽象成了事件处理器，而事件处理器之间通过事件相互关联。 每种事件处理器处理一种类型的事件，同时根据需要触发另外一种事件

 

相比于基于函数调用的编程模型，这种编程模型具有异步、并发等特点，更加高效，因此更适合大型分布式系统

 

六. 状态机库

 

状态机由一组状态组成，这些状态分为三类：初始状态、中间状态和最终状态。状态机从初始状态开始运行，经过一系列中间状态后，到达最终状态并退出。在一个状态机中，每个状态都可以接收一组特定事件，并根据具体的事件类型转换到另一个状态。当状态机转换到最终状态时，则退出

 

yarn状态转换方式

 

yarn中，每种状态转换由一个四元组表示，分别是转换前状态(prestate)、转换后状态(poststate)、事件(event)和回调函数(hook)

 

yarn定义了三种状态转换方式 ： 

 

(1) 一个初始状态、一个最终状态、一种事件。该方式表示状态机在prestate状态下，接收到event事件后，执行函数状态转移函数hook，并在执行完成后将当前状态转换为poststate

 

　　　　　　　　　　　　　　初始状态:最终状态:事件=1:1:1

 

(2) 一个初始状态、多个最终状态、一种事件。该方式表示状态机在prestate状态下，接收到event事件后，执行函数状态转移函数hook，并将当前状态转移为hook的返回值所表示的状态

 

　　　　　　　　　　　　　　　初始状态:最终状态:事件=1:n:1

 

(3) 一个初始状态、一个最终状态、多个事件。该方式表示状态机在prestate状态下，接收到event1、event2和event3中的任何一个事件，将执行函数状态转移函数hook，并在执行完成后将当前状态转换成poststate

 

　　　　　　　　　　　　　初始状态:最终状态:事件=1:1:n

 

状态机类

 

yarn自己实现了一个非常简单的状态机库(位于包org.apache.hadcop.yarn.state中)。yarn对外提供了一个状态机工厂statemachinefactory，它提供多种addtransition方法供用户添加各种状态转移，一旦状态机添加完毕后，可通过调用installtopology完成一个状态机的构建

 

我每天会写文章记录大数据技术学习之路,另外我自己整理了些大数据的学习资料,目前全部放在我的公众号"smallbird技术分享",加入我们一起学习交流,并且回复'分享'会有大数据资源惊喜等着你~