How MapReduce Works（转）

原文地址：http://www.cnblogs.com/ggjucheng/archive/2012/04/23/2465820.html

1、从Map到Reduce

MapReduce实际上是分治算法的一种实现，其处理过程亦和用管道命令来处理十分类似，一些简单的文本字符的处理甚至也可使用Unix的管道命令来替代，从处理流程的角度来看大概以下：

cat input | grep | sort | uniq -c | cat > output

# Input -> Map -> Shuffle & Sort -> Reduce -> Output

简单的流程图以下：

对于Shuffle，简单地说就是将Map的输出经过必定的算法划分到合适的Reducer中进行处理。Sort固然就是对中间的结果进行按key排序，由于Reducer的输入是严格要求按key排序的。

Input->Map->Shuffle&Sort->Reduce->Output只是从宏观的角度对MapReduce的简单描述，实际在MapReduce的框架中，即从编程的角度来看，其处理流程是Input->Map->Sort->Combine->Partition->Reduce->Output。用以前的对温度进行统计的例子来说述这些过程。

Input Phase

输入的数据须要以必定的格式传递给Mapper的，格式有多种，如TextInputFormat、DBInputFormat、SequenceFileInput等等，可使用JobConf.setInputFormat来设置，这个过程还应该包括对输入的数据进行任务粒度划分(split)而后再传递给Mapper。在温度的例子中，因为处理的都是文本数据，输入的格式使用默认的TextInputFormat便可。

Map Phase

对输入的key、value对进行处理，输出的是key、value的集合，即map (k1, v1) -> list(k2, v2)，使用JobConf.setMapperClass设置本身的Mapper。在例子中，将（行号、温度的文本数据）做为key/value输入，通过处理后，从温度的文件数据中提取出日期中的年份和该日的温度数据，造成新的key/value对，最后以list(年,  温度)的结果输出，如[(1950, 10), (1960, 40), (1960, 5)]。

Sort Phase

对Mapper输出的数据进行排序，能够经过JobConf.setOutputKeyComparatorClass来设置本身的排序规则。在例子中，通过排序以后，输出的list集合是按年份进行排序的list(年, 温度)，如[(1950, 10), (1950, 5), (1960, 40)]。

Combine Phase

这个阶段是将中间结果中有相同的key的<key, value>对合并成一对，Combine的过程与Reduce很类似，使用的甚至是Reduce的接口。经过Combine可以减小<key, value>的集合数量，从而减小网络流量。Combine只是一个可选的优化过程，而且不管Combine执行多少次(>=0)，都会使Reducer产生相同的输出，使用JobConf.setCombinerClass来设置自定义的Combine Class。在例子中，假如map1产生出的结果为[(1950, 0), (1950, 20), (1950, 10)]，在map2产生出的结果为[(1950, 15), (1950, 25)]，这两组数据做为Reducer的输入并通过Reducer处理后的年最高温度结果为(1950, 25)，然而当在Mapper以后加了Combine（Combine先过滤出最高温度），则map1的输出是[(1950, 20)]和map2的输出是[(1950, 25)]，虽然其余的三组数据被抛弃了，可是对于Reducer的输出而言，处理后的年最高温度依然是(1950, 25)。

Partition Phase

把Mapper任务输出的中间结果按key的范围划分红R份（R是预先定义的Reduce任务的个数），默认的划分算法是”(key.hashCode() & Integer.MAX_VALUE) % numPartitions”，这样保证了某一范围的key必定是由某个Reducer来处理，简化了Reducer的处理流程，使用JobConf.setPartitionClass来设置自定义的Partition Class。在例子中，默认就天然是对年份进行取模了。

Reduce Phase

Reducer获取Mapper输出的中间结果，做为输入对某一key范围区间进行处理，使用JobConf.setReducerClass来设置。在例子中，与Combine Phase中的处理是同样的，把各个Mapper传递过来的数据计算年最高温度。

Output Phase

Reducer的输出格式和Mapper的输入格式是相对应的，固然Reducer的输出还能够做为另外一个Mapper的输入继续进行处理。

2、Details of Job Run

上面只是从task运行中描述了Map和Reduce的过程，实际上当从运行”hadoop jar”开始还涉及到不少其余的细节。从整个Job运行的流程来看，以下图所示：

从上图能够看到，MapReduce运行过程当中涉及有4个独立的实体：

1. Client，用于提交MapReduce job。

2. JobTracker，负责协调job的运行。

3. TaskTrackers，运行 job分解后的多个tasks，task主要是负责运行Mapper和Reducer。

4. Distributed filesystem，用于存储上述实体运行时共享的job文件（如中间结果文件）。

Job Submission

当调用了JobClient.runJob()以后，Job便开始被提交了，在Job提交这个步骤中，经历了如下的过程：

1. Client向JobTacker申请一个新的job ID（step 2），job ID形如job_200904110811_0002的格式，是由JobTracker运行当前的job的时间和一个由JobTracker维护的自增计数（从1开始）组成的。

2. 检查job的output specification，好比输出目录是否已经存在（存在则抛异常）、是否有权限写等等。

3. Computes the input splits for the job，这些input splits就是做为Mapper的输入。

4. Copies the resources needed to run the job, including the job JAR file, the configuration file and the computed input splits, to the jobtracker’s filesystem in a direcotry named after the job ID（step 3）。

5. Tells the jobtracker that the job is ready for execution（step 4）。

Job Initialization

当JobTracker收到Job提交的请求后，将job保存在一个内部队列，并让Job Scheduler处理并初始化。初始化涉及到建立一个封装了其tasks的job对象，并保持对task的状态和进度的根据（step 5）。当建立要运行的一系列task对象后，Job Scheduler首先开始从文件系统中获取由JobClient计算的input splits（step 6），而后再为每一个split建立map task。

Task Assignment

TaskTrackers会使用一个简单的loop为按期向JobTracker发送heartbeat调用，发送的间隔时间大约5秒，通常取决于集群服务器的规模和繁忙程度以及网络拥挤程度。这个heartbeat一方面是告知JobTracker当前TaskTracker处于live状态，同时是用于JobTracker和TaskTracker进行通讯，TaskTracker会根据heartbeat的返回值来执行必定的操做（step 7）。

To choose a reduce task the JobTracker simply takes the next in its list of yet-to-be-run reduce tasks, since there are no data locality considerations. For a map task, however, it takes account of the TaskTracker’s network location and picks a task whose input splits is as close as possible to the tasktracker. In the optimal case, the task is data-local, that is , running on the same node that the split resides on. Alternatively, the task may be rack-local: on the same rack, but not the same node, as the split.

Task Execution

当TaskTrack被分配到一个task以后，接下来就是运行这个task。首先，它会须要的job JAR文件从shared filesystem拷贝到local filesystem，而后建立一个working direcotry并un-jars拷贝的JAR文件到该directory，最后就建立一个TaskRunner对象运行task。

TaskRunner在运行的时候是启动了一个新的JVM来run each task（step 10），这样是为了防止在用户自定义的Mapper出现异常令JVM挂了，从而连累到TaskTracker。TaskRunner子进程会使用umbilical接口和TaskTracker通讯并每隔几秒向TaskTracker汇报进度。

对于使用Streaming和Pipes方式来建立的Mapper，也是做为TaskTracker的子进程来运行的。Streaming是使用标准输入输出来通讯，而Pipes是使用socket来进行通讯，以下图：

Progress and Status Updates

进度和状态是经过heartbeat来更新和维护的。来对于Map Task，进度就是已处理数据和全部输入数据的比例。对于Reduce Task，状况就有点复杂，包括3部分，拷贝中间结果文件、排序、Reduce调用，每部分占1/3。

Job Completion

当Job完成后，JobTracker会收一个Job Complete的通知，并将当前的Job状态更新为Successful，同时JobClient也会轮循获知提交的Job已经完成，将信息显示给用户。最后，JobTracker会清理和回收该Job的相关资源，并通知TaskTracker进行相同的操做（好比删除中间结果文件）。