hadoop2.0 yarn 之 mapreduce on yarn (MRv2) 基于hadoop2.2.0

java高级交流群：37341439 欢迎所有学习hadoop2.2.0的朋友加入讨论交流，分享实战经验！我们群无下限哦~~

mapreduce on yarn (MRv2)

mapreduce on yarn 与MRv1在编程模型和数据处理引擎方面的实现是一样的，唯一的区别是运行时环境，mapreduce on yarn的运行时环境由 yarn 和applicationmaster构成。

编程模型：mapreduce的一些api

数据处理引擎：

说明：从MapReduce 自身的命名特点可以看出， MapReduce 由两个阶段组成：Map 和

Reduce。用户只需编写map() 和 reduce() 两个函数，即可完成简单的分布式程序的设计。

map() 函数以key/value 对作为输入，产生另外一系列 key/value 对作为中间输出写入本地

磁盘。 MapReduce 框架会自动将这些中间数据按照 key 值进行聚集，且key 值相同（用户可

设定聚集策略，默认情况下是对 key 值进行哈希取模）的数据被统一交给 reduce() 函数处理。

reduce() 函数以key 及对应的value 列表作为输入，经合并 key 相同的value 值后，产

生另外一系列 key/value 对作为最终输出写入HDFS

hello world --WordCount

Map部分如下：

// key: 字符串偏移量

// value: 一行字符串内容

map(String key, String value) ：

// 将字符串分割成单词

words = SplitIntoTokens(value);

for each word w in words:

EmitIntermediate(w, "1");

Reduce 部分如下：

// key: 一个单词

// values: 该单词出现的次数列表

reduce(String key, Iterator values):

int result = 0;

for each v in values:

result += StringToInt(v);

Emit(key, IntToString(result));

用户编写完MapReduce 程序后，按照一定的规则指定程序的输入和输出目录，并提交
到Hadoop 集群中。作业在Hadoop 中的执行过程如图所示。Hadoop 将输入数据切分
成若干个输入分片（input split，后面简称split），并将每个split 交给一个Map Task 处理；
Map Task 不断地从对应的split 中解析出一个个key/value，并调用map() 函数处理，处理完
之后根据Reduce Task 个数将结果分成若干个分片（partition）写到本地磁盘；同时，每个
Reduce Task 从每个Map Task 上读取属于自己的那个partition，然后使用基于排序的方法将
key 相同的数据聚集在一起，调用reduce() 函数处理，并将结果输出到文件中

map task (taskAttempt)

Map Task 执行过程如图2-7 所示。由该图可知，Map Task 先将对应的split 迭代解析成一
个个key/value 对，依次调用用户自定义的map() 函数进行处理，最终将临时结果存放到本地
磁盘上，其中临时数据被分成若干个partition，每个partition 将被一个Reduce Task 处理。

reduce task(taskAttempt)

Reduce Task 执行过程如图2-8 所示。该过程分为三个阶段①从远程节点上读取 Map

Task 中间结果（称为“Shuffle 阶段 ”）；②按照key 对 key/value 对进行排序（称为“Sort 阶

段”）；③依次读取 <key, value list>，调用用户自定义的reduce() 函数处理，并将最终结果

存到HDFS 上（称为 “Reduce 阶段” ）。

运行时环境：

yarn和MRAppMaster

说明：MRAppMaster是mapreduce的applicationMaster的实现，它使得MapReduce应用程序可以直接运行于yarn上，在yarn中，MRAppMaster负责管理MapReduce作业的生命周期，包括作业管理，资源申请与再分配，container启动与释放，作业恢复等。

MRAppMaster工作流程：

     按照作业大小不同，提供了3种作业运行模式

•           本地模式
•           Uber模式：为了降低小作业延时，yarn专门对小作业运行方式进行了优化，对于小作业而言，MRAppMaster无需再为每个任务分别申请资源，而是让其重用一个container，并按照先map task后reducetask的运行方式穿行执行每个任务，在yarn如果一个mapreduce作业同时满足以下条件，则认为是小作业，可运行在Uber模式下
• • Map Task数目不超过Mapreduce.job.ubertask.maxmaps(默认是9)
  • Reduce task数目不超过mapreduce.job.ubertask.maxmaps(默认是1)
  • 输入文件大小不超过mapreduce.job.ubertask.maxbytes(默认是一个block大小)
  • map task和reduce task需要的资源量不超过MRAppMaster可使用的资源量
  • 另外，由于链式作业会并发执行不同资源需求的map task和reduce task，因此不允许在uber模式下
•           Non-Uber模式：在大数据环境下，uber运行模式通常只能覆盖到一小部分作业，而对于其他大多数作业仍将运行在Non-Uber模式下，MRAppMaster将一个作业的map task和reduce task分为四种状态：
• • pending:刚启动但尚未向resourcemanager发送资源请求
  • scheduled：已经向resourceManager发送资源请求，但尚未分配到资源
  • assigned：已经分配到了资源且正在运行
  • completed已经运行完成。

map task 生命周期：scheduled-assigned-completed

reduce task 生命周期：pending-scheduled-assigned-completed

由于reduce task的执行需要依赖于map task的输出结果，因此，为避免reduce task过早启动造成资源利用率底下，MRAppMaster让刚启动的reduce处于pending状态，以便能够根据map task的运行情况决定是否对其进行调度。

在yarn之上运行mapreduce作业需要解决2个关键问题：

•      如何确定reduce task启动时机：由于yarn中没有map slot和reduce slot的概念，且resourcemanager也不知道map task和reduce task之间的依赖关系，因此MRAppMaster自己需要设计资源申请策略以防止因reduce task过早启动照成资源利用率低下和map task因分配不到资源而饿死。MRAppMaster在MRv1原有策略（map task完成数目达到一定比例后才允许启动reduce task）基础上添加了更为严格的资源控制策略和抢占策略
• • mapreduce.job.reduce.slowstart.completedmaps:当map task完成的比例达到改制后才会为reduce task申请资源，默认是0.05
  • yarn.app.mapreduce.am.job.reduce.rampup.limit:在maptask完成之前，最多启动reduce task比例，默认是0.5
  • yarn.app.mapreduce.am.job.reduce.preemption.limit:当map task需要资源但暂时无法获取资源（比如reduce task运行过程中，部分map task因结果丢失需重算）时，为了保证至少一个map task可以得到资源，最多可以抢占reduce task比例，默认是0.5
•      如何完成shuffle功能：按照mapreduce的执行逻辑，shuffle http server应该分布到各个节点上，以便能够支持各个reduce task远程复制数据，然而，由于shuffle是mapreduce框架中特有的一个处理流程，从设计上来说，不应该将它直接嵌到yarn的某个组件（nodemanager）中。在yarn中，nodemanager作为一种组合服务模式，允许动态加载应用程序临时需要的附属服务，利用这一特性，yarn将shuffle http server组装成一种服务，以便各个nodemanager启动时加载它。

 原帖地址：http://blog.csdn.NET/jiushuai/article/details/17733581