MapReduce扩展：应用程序如何运行于Hadoop Yarn之上

1. 背景

 

“应用程序运行于Hadoop Yarn之上”的需求来源于微博运维数据平台中的调度系统，即调度系统中的任务须要运行于Hadoop Yarn之上。这里的应用程序能够简单理解为一个普通的进程（这里特指Java进程），调度系统中的任务执行实际也是一个进程的运行过程，这里咱们不讨论为何调度系统中的任务（进程）须要运行于Hadoop Yarn之上，仅仅讨论如何使得一个应用程序（进程）能够运行于Hadoop Yarn之上。

 

应用程序（进程）须要运行于Hadoop Yarn之上，有三种可选的实现方案：

 

（1）扩展实现Yarn Application的两个组件：Yarn Client、ApplicationMaster；

（2）重用已有的计算框架，如：MapReduce、Spark；

（3）借助于开源框架，如：Apache Twill；

 

咱们的实现方案最终肯定为重用已有的计算框架：MapReduce，主要是基于如下几个因素考虑的：

 

（1）Yarn Application的Yarn Client、ApplicationMaster的扩展实现过程异常复杂，须要对Hadoop Yarn有很是深刻的了解，对于咱们的目标而言这种方式过于“大材小用”，毕竟咱们只是但愿一个普通的Java进程能够运行于Hadoop Yarn之上便可；

（2）Apache Twill及相似的开源框架旨在简化Yarn Application（Yarn Client、ApplicationMaster）的实现过程，但目前均处于孵化器状态，不建议实际环境中使用；

（3）已有的计算框架中，MapReduce、Spark使用都比较普遍，而Spark使用Scala开发，不太适用于团队目前的工程背影及与现有系统整合，所以选取了使用Java开发的MapReduce；

 

实现方案肯定以后，咱们的目标再也不是“应用程序如何运行于Hadoop Yarn之上”，变为“一个普通的Java进程如何运行于Hadoop Yarn之上”，而MapReduce仅仅包含两种类型的任务：Map Tasks、Reduce Tasks，进一步思考以后，咱们获得最终目标：“一个普通的Java进程如何以一个MapReduce MapTask的形式运行于Hadoop Yarn之上”。

 

为何是MapTask，而不是ReduceTask？

 

每个MapTask、ReduceTask都是一个Java进程，宏观上看，ReduceTask须要运行于MapTask以后，即ReduceTask的运行必须依赖于全部的MapTask结束以后才能够运行。

 

若是是“一个普通的Java进程以一个MapReduce ReduceTask的形式运行于Hadoop Yarn之上”，则Hadoop Yarn之上至少须要两个进程：一个MapTask进程和一个ReduceTask进程，这与一个普通的Java进程的初衷是不符的，至关于以前只须要一个进程就能够完成的任务，如今至少须要两个进程。

 

MapReduce是能够不须要ReduceTask的，这能够经过设置Hadoop MapReduce属性“mapreduce.job.reduces”的值为0来实现。

 

2. 实现

 

2.1 YarnApplication

 

一个普通的Java进程如今至关于一个MapReduce MapTask进程，这个“普通的Java进程”的计算逻辑实际能够是多种多样的，这就要求咱们须要把MapReduce MapTask进程看做是一个“容器”，运行于其中的应用能够有各类各样的计算逻辑，只有这样MapReduce MapTask的进程才能够等价于一个普通的Java进程，虽然它包含不少额外的执行过程（与应用计算逻辑无关的）。

 

根据以往的经验，“容器”中的应用一般须要实现特定的接口（Interface或Abstract class），为此咱们特地设计了一个抽象类：YarnApplication，用于表示“容器”中的应用。

 

 

属性

 

context：MapReduce MapTask Context（org.apache.hadoop.mapreduce.Mapper.Context<LongWritable, Text, LongWritable, Text>），即：MapReduce Mapper环境上下文，主要用于获取Hadoop MapReduce配置属性值；

 

方法

 

setContext：设置MapReduce Mapper Context（环境上下文）；

 

execute：应用的计算逻辑以方法的形式封装至方法execute，该方法能够接受任意个字符串形式的参数；

 

简而言之，一个普通的Java进程若是想运行于Yarn之上，如今仅仅须要实现本身特有的SpecificApplication，SpecificApplication须要继承自YarnApplication，并重写其中的execute方法，用于表示应用的计算逻辑，而后由“容器”（MapReduce MapTask）负责完成应用的执行过程，即：SpecificApplication execute的方法调用。

 

代码参考： https://github.com/Leaderman/scheduler-on-yarn/blob/master/src/main/java/com/weibo/dip/yarnscheduler/app/YarnApplication.java

 

2.2 AppMapper

 

AppMapper是MapReduce Mapper的一个具体实现，它的功能就是2.1中说起的“容器”，用于完成多种多样的YarnApplication的执行过程。

 

 

属性

 

appClass：既然“容器”能够支持多种多样的YarnApplication的执行，那么某一个具体的“容器实例”（即：AppMapper Task）启动时，须要知道具体执行哪个YarnApplication实例；appClass用于保存YarnApplication实例的彻底限定类名，它的具体值能够在Hadoop MapRedcue启动时经过参数“app.class”进行指定；

 

args：用于表示YarnApplication execute方法执行时须要传递的参数，它的具体值能够在Hadoop MapReduce启动时经过参数“app.args”进行指定；

 

方法

 

setup：AppMapper实例的初始化过程当中获取appClass与args的具体值；

 

 

map：AppMapper实例的具体执行过程，负责完成YarnApplication execute的调用过程；

 

 

具体的执行过程分为三步：

 

（1）经过反射加载具体的YarnApplication实现类，并建立相应的实例app；

（2）实例app设置相应的Mapper环境上下文context；

（3）实例app执行execute方法；

 

cleanup：暂时没有使用；

 

这里有一点须要额外注意：AppMapper map没有处理任何的数据输入输出。

 

代码参考： https://github.com/Leaderman/scheduler-on-yarn/blob/master/src/main/java/com/weibo/dip/yarnscheduler/mapreduce/AppMapper.java

 

2.3 AppInputFormat

 

AppMapper仅仅须要一个Map Task，所以InputSplit的数目为1；AppMapper map仅仅执行一次，意味着InputSplit的记录数目为1；这样的需求若是使用TextInputFormat，则要求咱们必须在HDFS上存储一个文本文件，这个文本文件仅仅包含有一行文本，对于咱们的场景而言，太过烦琐，所以咱们设计实现了专用的InputFormat：AppInputFormat。

 

InputFormat须要有两个核心组件组成：InputSplit和RecordReader，AppInputFormat也不例外，以下：

 

 

如上所述，AppMapper map没有处理任何的数据输入，所以，AppInputFormat须要的InputSplit能够是“虚拟”的，InputSplit中的记录也能够是“虚拟”的。

 

2.3.1 AppInputSplit

 

AppInputSplit就是一个“虚拟”的InputSplit，它没有引用或关联任何实际的数据。

 

 

方法

 

getLength：AppInputSplit是“虚拟”的，没有引用或关联任何实际的数据，所以数据长度为0；

 

getLocations：AppInputSplit是“虚拟”的，没有引用或关联任何实际的数据，不须要考虑数据本地性的问题，所以仅返回一个“localhost”便可；

 

write、readFields：AppInputSplit不包括任何实例属性，由于序列化方法（write）和反序列化方法（readFields）为空便可；

 

2.3.2 AppRecordReader

 

虽然AppInputSplit是“虚拟”的，但它依然须要一个对应的RecordReader，且这个RecordReader须要可以从AppInputSplit中“读取”到一条记录，不然AppMapper map方法没法获得执行。

 

 

属性

 

progress：表示AppInputSplit的处理进度，由于AppInputSplit仅仅包含一条“虚拟”记录，所以progress只有两个值：0.0和1.0，初始值为0.0；

key：表示AppInputSplit中的那条“虚拟”记录的KEY；

value：表示AppInputSplit中的那条“虚拟”记录的VALUE;

 

方法

 

initialize：初始化AppRecordReader实例，为空便可；

 

nextKeyValue：用于表示AppInputSplit中是否仍有记录能够读取，若是实例变量key和value均为null，表示有一条记录能够读取；不然表示读取完成，progress置为1.0；

 

getCurrentKey：若是nextKeyValue()方法返回值为true，表示有一条记录能够读取，getCurrentKey()返回这条记录的KEY，由于只有一条记录，咱们这里将KEY设置为-1，并保存至实例变量key；

 

getCurrentValue：若是nextKeyValue()方法返回值为false，表示有一条记录能够读取，getCurrentValue()返回这条记录的VALUE，由于只有一条记录，咱们这里将VALUE设置为“APP Record”，并保存至实例变量value；

 

getProgress：直接返回实例变量progress的值便可；

 

close：没有使用，为空便可；

 

这里有一点须要注意，AppRecordReader中并无使用到AppInputSplit，这是由于AppInputSplit及其中的记录均可以理解为是“虚拟”的，AppRecordReader只须要可以“读取”到一条记录便可，至于这条记录是否是实际包含在AppInputSplit中是可有可无的。

 

2.3.3 AppInputFormat

 

 

方法

 

getSplits：AppMapper仅仅须要一个MapTask，所以只须要以数组的形式返回一个AppInputSplit实例便可；

 

createRecordReader：返回一个AppRecordReader实例便可；

 

代码参考： https://github.com/Leaderman/scheduler-on-yarn/blob/master/src/main/java/com/weibo/dip/yarnscheduler/mapreduce/AppInputFormat.java

 

2.4 AppOutputFormat

 

整个Hadoop MapReduce的运行过程当中仅仅只有一个AppMapper Task，而这个惟一的AppMapper Task没有任何输出，所以咱们须要一个“空”的OutputFormat：AppOutputFormat。

 

OutputFormat须要有两个核心组件组成：RecordWriter和OutputCommitter，AppOutputFormat也不例外，以下：

 

 

2.4.1 AppRecordWriter

 

 

如上所述，由于没有任何数据输出，因此AppRecordWriter中的全部方法为空便可。

 

2.4.2 AppOutputCommitter

 

 

OutputFormat使用OutputCommitter用于“提交”Task的输出，由于没有任何数据输出，因此AppOutputCommitter中的全部方法为空便可。

 

2.4.3 AppOutputFormat

 

 

方法

 

getRecordWriter：返回一个AppRecordWriter实例便可；

 

checkOutputSpecs：没有任何数据输出，为空便可；

 

getOutputCommitter：返回一个AppOutputCommitter实例便可；

 

代码参考： https://github.com/Leaderman/scheduler-on-yarn/blob/master/src/main/java/com/weibo/dip/yarnscheduler/mapreduce/AppOutputFormat.java

 

2.5 AppMapReduce

 

 

AppMapReduce继承类Configured，并实现接口Tool，至关于MapReduce的一个驱动，可使用如下的方式来运行：

 

 

注：ToolRunner、Configured、Tool是Hadoop MapReduce提供的工具类，方便咱们运行Hadoop MapReduce。

 

代码参考： https://github.com/Leaderman/scheduler-on-yarn/blob/master/src/main/java/com/weibo/dip/yarnscheduler/mapreduce/AppMapReduce.java

 

综上所述，YarnApplication、AppMapper、AppInputFormat（AppInputSplit、AppRecordReader）、AppOutputFormat（AppRecordWriter、AppOutputCommitter）、AppMapReduce之间的类关系以下图：

 

 

3. 示例

 

假设咱们一个“HelloWorld”的应用，以一个普通的Java进程的实现方式以下：

 

 

若是想让这个“HelloWorld”的应用运行于Hadoop Yarn之上，咱们须要实现一个HelloWorldApplication，它继承自YarnApplication，并重写其中的execute方法：

 

 

HelloWorldApplication的运行须要一个“驱动程序”：

 

 

这个“驱动程序”的运行方式以下：

 

java -cp conf/:target/scheduler-on-yarn-0.0.1-SNAPSHOT.jar:target/scheduler-on-yarn-0.0.1-SNAPSHOT-lib/* com.weibo.dip.yarnscheduler.example.HelloWorldApplicationExecutor -conf /etc/hadoop-offline/conf/core-site.xml -conf /etc/hadoop-offline/conf/hdfs-site.xml -conf /etc/hadoop-offline/conf/mapred-site.xml -conf /etc/hadoop-offline/conf/yarn-site.xml -D mapreduce.job.queuename=hive -D app.class=com.weibo.dip.yarnscheduler.app.HelloWorldApplication -D app.args=abc,def,ghi

 

参数说明：

 

-conf /etc/hadoop-offline/conf/core-site.xml

-conf /etc/hadoop-offline/conf/hdfs-site.xml

-conf /etc/hadoop-offline/conf/mapred-site.xml

-conf /etc/hadoop-offline/conf/yarn-site.xml

 

这几个参数用于指定Hadoop集群的配置文件；

 

-D mapreduce.job.queuename=hive

 

用于指定Hadoop MapReduce运行于Hadoop Yarn中的哪一个队列；

 

-D app.class=com.weibo.dip.yarnscheduler.app.HelloWorldApplication

 

用于指定YarnApplication的具体实现类，此处为HelloWorldApplication；

 

-D app.args=abc,def,ghi

 

用于指定HelloWorldApplication运行时所须要的参数，多个参数以“,”进行分隔。

 

代码参考：

 

https://github.com/Leaderman/scheduler-on-yarn/blob/master/src/main/java/com/weibo/dip/yarnscheduler/example/HelloWorldMain.java

https://github.com/Leaderman/scheduler-on-yarn/blob/master/src/main/java/com/weibo/dip/yarnscheduler/app/HelloWorldApplication.java

https://github.com/Leaderman/scheduler-on-yarn/blob/master/src/main/java/com/weibo/dip/yarnscheduler/example/HelloWorldApplicationExecutor.java