《深入解析YARN架构设计与实现原理》第3章 YARN基础库【3.4】

javazx · 发表于 2017-4-17 13:58:22

3.4 服务库与事件库
本节介绍服务库和事件库。
3.4.1 服务库
对于生命周期较长的对象， YARN采用了基于服务的对象管理模型对其进行管理，该模型主要有以下几个特点。
❑将每个被服务化的对象分为4个状态： NOTINITED（被创建）、 INITED（已初始化）、 STARTED（已启动）、
STOPPED（已停止）。
❑任何服务状态变化都可以触发另外一些动作。
❑可通过组合的方式对任意服务进行组合，以便进行统一管理。
YARN中关于服务模型的类图（位于包org.apache.hadoop.service中）如图3-13所示。在这个图中，我们可以看到，所有的服务
对象最终均实现了接口Service，它定义了最基本的服务初始化、启动、停止等操作，而AbstractService类提供了一个最基本的
Service实现。 YARN中所有对象，如果是非组合服务，直接继承AbstractService类即可，否则需继承CompositeService。比如，对于
ResourceManager而言，它是一个组合服务，它组合了各种服务对象，包括ClientRMService、 ApplicationMasterLauncher、
ApplicationMasterService等。
图3-13 YARN中服务模型的类图
在YARN中， ResourceManager和NodeManager属于组合服务，它们内部包含多个单一服务和组合服务，以实现对内部多种服
务的统一管理。
3.4.2 事件库
YARN采用了基于事件驱动的并发模型，该模型能够大大增强并发性，从而提高系统整体性能。为了构建该模型， YARN将
各种处理逻辑抽象成事件和对应事件调度器，并将每类事件的处理过程分割成多个步骤，用有限状态机表示。 YARN中的事件处
理模型可概括为图3-14所示。
图3-14 YARN的事件处理模型
整个处理过程大致为：处理请求会作为事件进入系统，由中央异步调度器（ Async-Dispatcher）负责传递给相应事件调度器
（ Event Handler）。该事件调度器可能将该事件转发给另外一个事件调度器，也可能交给一个带有有限状态机的事件处理器，其
处理结果也以事件的形式输出给中央异步调度器。而新的事件会再次被中央异步调度器转发给下一个事件调度器，直至处理完成
（达到终止条件）。
在YARN中，所有核心服务实际上都是一个中央异步调度器，包括ResourceManager、 NodeManager、
MRAppMaster（ MapReduce应用程序的ApplicationMaster）等，它们维护了事先注册的事件与事件处理器，并根据接收的事件类型
驱动服务的运行。
YARN中事件与事件处理器类的关系（位于包org.apache.hadoop.yarn.event中）如图3-15所示。当使用YARN事件库时，通常先
要定义一个中央异步调度器AsyncDispatcher，负责事件的处理与转发，然后根据实际业务需求定义一系列事件Event与事件处理器
EventHandler，并注册到中央异步调度器中以实现事件统一管理和调度。以MRAppMaster为例，它内部包含一个中央异步调度器
AsyncDispatcher，并注册了TaskAttemptEvent/TaskAttemptImpl、 TaskEvent/TaskImpl、 JobEvent/JobImpl等一系列事件/事件处理器，由
中央异步调度器统一管理和调度。
服务化和事件驱动软件设计思想的引入，使得YARN具有低耦合、高内聚的特点，各个模块只需完成各自功能，而模块之间
则采用事件联系起来，系统设计简单且维护方便。
图3-15 事件与事件处理器
3.4.3 YARN服务库和事件库的使用方法
为了说明YARN服务库和事件库的使用方法，本小节介绍一个简单的实例，该实例可看做MapReduce
ApplicationMaster（ MRAppMaster）的简化版。该例子涉及任务和作业两种对象的事件以及一个中央异步调度器。步骤如下。
1）定义Task事件。
public class TaskEvent extends AbstractEvent<TaskEventType> {
private String taskID; //Task ID
public TaskEvent(String taskID, TaskEventType type) {
super(type);
this.taskID = taskID;
} p
ublic 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;
} p
ublic String getJobId() {
return 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 = new String(jobID + "_task_" + i);2 p# \- C7 ^; @& y# `6 I
}. B4 o: D, e2 K" z
} p# v) ^6 M  V9 {$ L& \
ublic void serviceInit(final Configuration conf) throws Exception {
& \! y4 R2 L' }/ Rdispatcher = new AsyncDispatcher();//定义一个中央异步调度器4 r- Y, W+ B/ L% s8 ~% ?
//分别注册Job和Task事件调度器8 Z' s, j3 g8 c  E1 ~6 T; Z: l
dispatcher.register(JobEventType.class, new JobEventDispatcher());
4 ?4 ]- o  Q: w0 Vdispatcher.register(TaskEventType.class, new TaskEventDispatcher());
  \4 I; h* i5 a/ `; A8 G" ZaddService((Service) dispatcher);
; G7 z+ {; J& y1 _: Z' Gsuper.serviceInit(conf);
6 k$ w3 E( \3 [* ]9 W} p
( n2 I6 Y7 g2 Bublic Dispatcher getDispatcher() {
5 ~. E1 e2 M. j. g5 r7 \4 {return dispatcher;
: U/ [: G! w9 w4 V& c- P} p0 y' D, Y/ A0 v) Z1 ]1 M" n# E
rivate class JobEventDispatcher implements EventHandler<JobEvent> {
5 [8 S4 Q$ Y4 M( N; v0 e@Override' ]2 j+ f/ R" p0 [5 j' v
public void handle(JobEvent event) {- ^, z( m+ s8 J, \: C
if(event.getType() == JobEventType.JOB_KILL) {
+ d& y- t* J# L% i7 f' JSystem.out.println("Receive JOB_KILL event, killing all the tasks");
7 D! o9 M9 v; N. M8 {( `& efor(int i = 0; i < taskNumber; i++) {% l: E' u5 Y7 e, S5 S! o
dispatcher.getEventHandler().handle(new TaskEvent(taskIDs,
) Y9 e4 T9 I& J7 g, B4 x' `$ FTaskEventType.T_KILL));+ e8 ?8 _) n: d# B( m
}  e: Z+ f7 b$ F) S0 F5 J* P
} else if(event.getType() == JobEventType.JOB_INIT) {
6 f6 W4 T- h& ?" |6 n: U" oSystem.out.println("Receive JOB_INIT event, scheduling tasks");2 x; F7 f. e% J& P! H$ W
for(int i = 0; i < taskNumber; i++) {
2 w, u+ t1 ]* t+ Ndispatcher.getEventHandler().handle(new TaskEvent(taskIDs,5 S9 X& z- @3 I; S
TaskEventType.T_SCHEDULE));- {9 z, f" [4 {7 g& U2 {2 E
}
6 d" M; R) C- R9 b( W}
" R3 z; x2 s4 I3 G! B}2 R  z6 }5 a; Y6 S: J
}p
3 ]8 \( T/ E, s- erivate class TaskEventDispatcher implements EventHandler<TaskEvent> {1 V3 n! i/ B9 Z
@Override4 v% `! F: Q2 E
public void handle(TaskEvent event) {
: g/ G$ a2 m+ c+ H& V! ^$ Q2 Pif(event.getType() == TaskEventType.T_KILL) {
) d3 u2 c$ ^& |. N9 b6 {System.out.println("Receive T_KILL event of task " + event.getTaskID());
9 |' x$ o+ X+ y% ^2 j} else if(event.getType() == TaskEventType.T_SCHEDULE) {7 s+ D2 A; _- X8 r
System.out.println("Receive T_SCHEDULE event of task " + event.getTaskID());
1 l, g; E* c/ K( l2 |8 B, ^# r9 E; l}
6 ]. \: Y1 x( H* u5 R7 e  ?& Z}0 r' A6 u5 K1 w  M4 s
}5 H% ^/ H( }& q) |6 F
}
3 ~1 g- ]* Z9 t3 i5 g' I& S' s4）测试程序。
% A( r! x" m8 _0 ~: b@SuppressWarnings("unchecked")0 l# `, h8 }  F4 ?1 _6 b
public class SimpleMRAppMasterTest {2 c, r: F# M: r
public static void main(String[] args) throws Exception {
+ D: d6 Z, B. j2 ~String jobID = "job_20131215_12";$ o6 `* p' N" R3 U, q
SimpleMRAppMaster appMaster = new SimpleMRAppMaster("Simple MRAppMaster", jobID, 5);
; G) @: M- q8 C5 Z5 YYarnConfiguration conf = new YarnConfiguration(new Configuration());
3 K5 l% i' ~" x0 |, H; KappMaster.serviceInit(conf);$ {6 [4 p( x' D9 L) n
appMaster.serviceStart();
  }3 B2 J( z3 L8 N) \* QappMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID,8 S. r" X( r" a7 u+ K
JobEventType.JOB_KILL));
! d; G2 E6 q' ~0 u* @appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID,1 b" ]. h' D* A8 k2 p' i
JobEventType.JOB_INIT));
: |; _: \9 i' M& H0 n# r}- x3 f3 u- [$ o
3.4.4 事件驱动带来的变化
; t8 r4 N7 z( `0 ]3 c$ l: v在MRv1中，对象之间的作用关系是基于函数调用实现的，当一个对象向另外一个对象传递信息时，会直接采用函数调用的9 j3 E6 g4 i. E2 J# G! K( X* ^
方式，且整个过程是串行的。比如，当TaskTracker需要执行一个Task时，将首先下载Task依赖的文件（ JAR包、二进制文件等、
% S3 c; K* r% C3 F* E+ t字典文件等）、然后执行Task。同时在整个过程中会记录一些关键日志，该过程可用图3-16描述。在整个过程中，下载依赖文件
- H- [" z% X3 }( n0 F$ D是阻塞式的，也就是说，前一个任务未完成文件下载之前，后一个新任务将一直处于等待状态，只有在下载完成后，才会启动一7 d9 t7 L  G% r$ t3 ^5 a/ i5 K% Y
个独立进程运行该任务。尽管后来MRv1通过启动过独立线程下载文件解决了该问题 [15] ，但这种方式不是在大系统中彻底解决3 c! m. B/ L7 F
问题之道，必须引入新的编程模型。$ I$ z/ Z" M, j. y
图3-16 基于函数调用的工作流程
& f' q  S# C8 e+ S0 `0 ~8 p基于函数调用的编程模型是低效的，它隐含着整个过程是串行、同步进行的。相比之下， MRv2引入的事件驱动编程模型则
2 Z8 @/ l' v: e是一种更加高效的方式。在基于事件驱动的编程模型中，所有对象被抽象成了事件处理器，而事件处理器之间通过事件相互关
; M6 M5 S. U" b: C7 m6 F, Z! B联。每种事件处理器处理一种类型的事件，同时根据需要触发另外一种事件，该过程如图3-17所示，当A需要下载文件时，只需
! j, ?) O) x5 V6 F向中央异步处理器发送一个事件即可（之后可以继续完成后面的功能而无须等待下载完成），该事件会被传递给对应的事件处理
& v# J2 U  {' A5 b# m器B，由B完成具体的下载任务。一旦B完成下载任务，便可以通过事件通知A。
$ _! P% [% S+ l! F图3-17 基于事件驱动的工作流程
0 p/ k" m& z" r  t' j- M: I相比于基于函数调用的编程模型，这种编程方式具有异步、并发等特点，更加高效，因此更适合大型分布式系统。
2 b5 R/ J+ ^/ Y8 T[15] 参见网址https://issues.apache.org/jira/browse/MAPREDUCE-2705。  + ]4 a1 j4 g8 l/ N( ~

9 X8 {# p- ?/ N3 _9 c3 K; Y3 r, o$ ]# f, W2 A* i( c

		自动登录	找回密码
密码			立即注册

《深入解析YARN架构设计与实现原理》第3章 YARN基础库【3.4】

相关帖子

宣传达人

突出贡献

优秀版主

荣誉管理

论坛元老