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
本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请
点击举报。
