关于Hadoop的MapReduce纯技术点文章

本文重点讲述Hadoop的整个MapReduce过程，不讲故事，不废话，重在描述每个环节。通过对google过来的文章一番苦读，我整了一些笔记，加入一些自己的看法，不一定全对，大家要区别对待。希望本文对想了解Hadoop的MapReduce的同学有一些帮助。

一． 使用Map/Reduce算法的目标
1) 能够将计算分布式处理
a) 当需要时，数据是总是可用的
b) 应用程序不用关心有多少计算机做服务
2) 提供高可靠性
a) 应用程序不关心机算机或者网络可能有临时或永久失效

二． 应用程序要做什么？
1) 定义Mapper和Reducer类和一个“启动”程序就可以完成整个过程
2) Mapper
a) 输入是一个key1,value1值对
b) 输出一个kye2,value2值对
3) Reducer
a) 输入一个key2和一个value2
b) 输出一个key3,value3的值对
4) 启动程序
a) 创建一个JobConf定义一个job
b) 提交JobConf到JobTracker并等待执行完毕

三． 应用程序数据流图
         详见附件图一
 
四． 输入和输出格式化
应用程序也要选择输入和输出格式，用来定义持久数据如何来读写。有接口可以定义。
1) 输入格式
a) 分隔输入数据决定输入到每个map task
b) 定义一个RecordReader来读key,value值对，传递到map task
2) 输出格式
a) 给出一个key,value值对和一个文件名，将reduce task的输出写到持久化存储中去

五． 输出排序
    应用程序能控制排序命令和通过OutputKeyComparator和Partitioner输出
1) OutputKeyComparator
a) 定义如何比较序列化键值
b) 默认OutputKeyComparator，但应该将此定义在一个应用程序里。
i. 写法如下Key.compareTo(key2)
2) Partitioner分配器
a) 给出一个map输出key和reduce的数量，选择一个reduce
b) 默认HashPartitioner，利用模计算来处理分配工作
i. key.hashCode % numReduces

六． 组合器
Combiners是一个带有合并多值到一个值的reduce的jobs的优化器。
1) 通常来看，combiner和reducer一样能通过map的输出来运行，只是输出到reducer之前运行。这也是符合移动计算要比移动数据的成本低的原则。
2) 例如，WordCount的mapper生成(word,count)和combiner和reducer生成每个词的合。
a) Input:“hi Kevin bye Kein”
b) Map output: (“hi”,1),(“Kevin”,1),(“bye”,1),(“Kevin”,1)
c) Combine output: (“Kevin”,2),(“bye”,1),(“hi”,1)

七． 过程通讯
1) 用一个普通的RPC实现
a) 容易变化/扩展
b) 定义成Java接口
c) 实现接口的服务对象
d) 客户端代理对象自动生成
2) 所有的消息在客户端发生
a) 防止循环进而死锁
3) 错误处理
a) 包括超时和通讯问题
b) 通过IOException发信号到客户端
c) 从来不发信号到服务器

八． Map/Reduce过程
1) 启动应用程序
a) 用户应用程序代码
b) 提交一个细节的Map/Reduce job
2) JobTracker
a) 处理所有的job
b) 生成所有任务决定的安排表
3) TaskTracker
a) 管理所有得到的节点任务
4) Task
a) 为一个指定的job运行一个单独的map或者reduce段
b) 任务从TaskTracker上获取

九． 执行过程交互图
         详见附件图二

十． Job控制流程
1) 应用程序启动器创建和提交job
2) JobTracker初始化job,创建FileSplits并将task加到队列中
3) 当前置任务完成后，TaskTracker每10秒种请求一个新的map或reduce任务
4) 随着任务运行，TaskTraker每10秒向JobTraker报告状态
5) 当job完成时，JobTracker告诉TaskTraker删除临时文件
6) 应用程序启动器通知job已经完成并且停止等待

十一． 应用程序启动器
1) 应用程序代码创建JobConf和参数集合
a) 定义Mapper,Reducer类
b) 定义InputFormat和OutFormat类
c) 如果需要的话，定义Combiner类
2) 写一个JobConf和应用程序jar到DFS和提交job到JobTracker
3) 可以立即退出或者等待job完成或者失败

十二． JobTracker
1) 处理JobConf并且创建一个InputFormat实例。调用getSplits方法生成map的输入
2) 创建一个JobInProgress对象和一个TaskInProgress“TIP”串和任务对象
a) JobInProgress 一个job的状态
b) TaskInProgress是一个工作段的状态
c) Task是做一个TIP尝试
d) 随着TaskTrackers 请求工作开始，他们被交给指定的task去执行

十三． TaskTracker
1) 所有任务
a) 创建TaskRunner
b) 从DFS复制job.jar和job.xml
c) 为本次任务本地化JobConf
d) 调用task.preare()
e) 在TaskTracker.Child下的一个新的JVM启动任务
f) 在log的info级别下抓取任务的输出
g) 处理任务状态的更新并且每10秒种送到JobTracker
h) 如果job被杀掉，同样杀掉task
i) 如果任务死掉或者完成，告之JobTracker

十四． TaskTracker for Reduces
1) 作为Reduces，task.prepare()为这个reduce抓取所有相关的map输出
2) 从TaskTracker的Jetty服务上，文件通过http抓取
3) 文件在平行线程上被抓取到，但是每个主机只有一个
4) 当抓取失败，一个回退方案被启用以抑制过载的TaskTracker
5) 抓取占用整个reduce进程的第一个33%进度

十五． Map Tasks
1) 用InputFormat对象从FileSplit创建一个RecordReader
2) 通过在FileSplit的 keys和values循环，并且供应给mapper
3) 如果没有combiner,一个用keys写成SequenceFile分配到每个reduce
4) 有combiner，框架缓存100000个keys和values,sorts,combines，并且将其写入SequenceFile分配到每个reduce

十六． Reduce Tasks:Sort
1) 排序
a) 占用33%到66%的reduce进度
b) 基本处理
i. 读100个(配置参数io.sort.mb)keys和values的值对到排序
ii. 在内存中排序
iii. 写到磁盘中
c) 合并
i. 读10个(io.sort.factor)文件并且合并到一个文件
ii. 重复许多次是必要的（100个文件二级，1000个文件三级,等等）

十七． Reduce Tasks:Reduce
1) 占用66%到100%的reduce进程
2) 用一个SequenceFile.Reader来读取排序的输入并且传递到reducer一个key，同时和相关值一起
3) 输入keys和values的值对被写成OutputFormat对象，通常是写到DFS的一个文件中去
4) 从reduce输出没有被采用，因此它是map输出keys的碎片

写完了，最好的理解办法，就是找个Hadoop例子逐条对应一下，如果有写得不对的地方，欢迎指正！