走向云计算之MapReduce原理和运行流程详解

一、概述

MapReduce的设计理念源自于Google的MapReduce论文（发表于2004年12月），Hadoop MapReduce是Google MapReduce克隆版。MapReduce采用”分而治之”的思想，把对大规模数据集的操作，分发给一个主节点管理下的各个分节点共同完成，然后通过整合各个节点的中间结果，得到最终结果。简单地说，MapReduce就是”任务的分解与结果的汇总”。

MapReduce1.0的架构如下：



在Hadoop1.x中，用于执行MapReduce任务的机器角色有两个：一个是JobTracker；另一个是TaskTracker，JobTracker是用于调度工作的，TaskTracker是用于执行工作的。一个Hadoop集群中只有一台JobTracker。

MapReduce2.0架构如下：



MapReduce2.0架构是在YARN架构的基础上运行，具体可参考上一篇关于YARN架构的介绍博文。

在分布式计算中，MapReduce框架负责处理了并行编程中分布式存储、工作调度、负载均衡、容错均衡、容错处理以及网络通信等复杂问题，把处理过程高度抽象为两个函数：map和reduce，map负责把任务分解成多个任务，reduce负责把分解后多任务处理的结果汇总起来。需要注意的是，用MapReduce来处理的数据集（或任务）必须具备这样的特点：待处理的数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全并行地进行处理。

二、MapReduce运行流程详解

MapReduce将作业的整个运行过程分为两个阶段：Map阶段Reduce阶段。

Map阶段由一定数量的Map Task组成，例如：

输入数据格式解析：InputFormat

输入数据处理：Mapper

本地规约：Combiner（相当于local reducer，可选）

数据分组：Partitioner

Reduce阶段由一定数量的Reduce Task组成，例如：

数据远程拷贝

数据按照key排序

数据处理：Reducer

数据输出格式：OutputFormat

通常我们把从Mapper输出数据到Reduce读取数据之间的过程称之为shuffle。

一个MapReduce的基本流程图如下：



现在来结合hadoop提供的WordCount程序来对其执行流程进行详解。源码如下：（针对本地环境对源码作了一些修改）

package com.kang;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

private final static IntWritable one = new IntWritable(1);
private Text word = new Text();

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
System.out.println("MapOut key:" + word.toString() + " value:" + one.toString());
context.write(word, one);
}
}
}

public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
System.out.println("vlaue:" + val.toString());
sum += val.get();
}
result.set(sum);
System.out.println("key:" + key.toString() + " valueSum:" + result.toString());
context.write(key, result);
}
}

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = new Job(conf, "word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path("hdfs://sparkproject1:9000/root/input/"));
FileOutputFormat.setOutputPath(job, new Path("hdfs://sparkproject1:9000/root/output/"));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

1、Hadoop相关数据类型

Hadoop提供了如下内容的数据类型，这些数据类型都实现了WritableComparable接口，以便用这些类型定义的数据可以被序列化进行网络传输和文件存储，以及进行大小比较。

BooleanWritable：标准布尔型数值
ByteWritable：单字节数值
DoubleWritable：双字节数
FloatWritable：浮点数
IntWritable：整型数
LongWritable：长整型数
Text：使用UTF8格式存储的文本
NullWritable：当<key,value>中的key或value为空时使用

2、代码详解

首先来看main函数

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = new Job(conf, "word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path("hdfs://sparkproject1:9000/root/input/"));
FileOutputFormat.setOutputPath(job, new Path("hdfs://sparkproject1:9000/root/output/"));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}

在MapReduce中，由Job对象负责管理和运行一个计算任务，并通过Job的一些方法对任务的参数进行相关的设置。此处设置了使用TokenizerMapper完成Map过程中的处理；使用IntSumReducer完成Combine和Reduce过程中的处理。还设置了Map过程和Reduce过程的输出类型：key的类型为Text，value的类型为IntWritable。任务的输出和输入路径则由FileInputFormat和FileOutputFormat分别设定。完成相应任务的参数设定后，即可调用job.waitForCompletion()方法执行任务。

split阶段



在进行map计算之前，mapreduce会根据输入文件计算输入分片（input split），每个输入分片（input split）针对一个map任务，输入分片（input split）存储的并非数据本身，而是一个分片长度和一个记录数据位置的数组。输入分片（input split）往往和HDFS的block（块）关系很密切，假如我们设定HDFS的块的大小是64MB，如果我们输入有三个文件，大小分别是3MB、65MB和127MB，那么mapreduce会把3MB文件分为一个输入分片（input split），65MB则是两个输入分片（input split），而127MB也是两个输入分片（input split）。换句话说我们如果在map计算前不做输入分片调整（例如合并小文件），那么就会有5个map任务将执行，而且每个map执行的数据大小不均，这个也是mapreduce优化计算的一个关键点。

InputFormat（默认TextInputFormat）阶段

将文件拆分成splits后，MapReduce框架会根据用户指定的InputFormat格式进行数据分割，框架默认使用TextInputFormat格式。框架默认将文件按行分割形成

<key,value>

对，这一步由MapReduce框架自动完成，其中偏移量（即key值）包括了回车所占的字符数。



map阶段

map阶段就是程序员编写好的map函数了，因此map函数效率相对好控制，而且一般map操作都是本地化操作也就是在数据存储节点上进行。本例的map函数如下：

public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

private final static IntWritable one = new IntWritable(1);
private Text word = new Text();

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
System.out.println("MapOut key:" + word.toString() + " value:" + one.toString());
context.write(word, one);
}
}
}

在MapReduce2.0中，Map过程需要继承org.apache.hadoop.mapreduce包中Mapper类，并重写其map方法。通过在map方法中添加两句把的key值和value值输出到控制台的代码，可以发现map方法中value值存储的是文本文件中的一行（以回车符为行结束标记），而key值为该行的首字母相对于文本文件的首地址的偏移量。然后StringTokenizer类将每一行拆分成为一个个的单词，并将

<word,1>

作为map方法的结果输出，其余的工作都交有MapReduce框架处理。

这部分的工作可以用下图表示：



shuffle阶段

将map的输出作为reduce的输入的过程就是shuffle。它包含了map端的combine阶段、排序阶段、Partitioner阶段，reduce端的排序阶段等。下面进行一一详述。

1、combine阶段

combiner阶段是程序员可以选择的，combiner其实也是一种reduce操作，因此我们看见WordCount类里是用reduce类来执行combine操作的。Combiner是一个本地化的reduce操作，它是map运算的后续操作，主要是在map计算出中间文件前做一个简单的合并重复key值的操作。例如我们对文件里的单词频率做统计，map计算时候如果碰到一个hadoop的单词就会记录为1，但是这个hadoop单词可能会出现n多次，那么map输出文件冗余就会很多。因此在reduce计算前先对相同的key做一个合并操作，那么文件会变少，这样就提高了宽带的传输效率。毕竟hadoop计算力宽带资源往往是计算的瓶颈也是最为宝贵的资源。

但是combiner操作是有风险的，使用它的原则是combiner的输入不会影响到reduce计算的最终输入。例如：如果计算只是求总数，最大值，最小值可以使用combiner。但是做平均值计算使用combiner的话，最终的reduce计算结果就会出错。

对于Hadoop而言，得到map方法输出的

<key,value>

对后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key至相同value值累加，得到Mapper的最终输出结果。当然我们可以把这个排序归入到combine过程中。

该阶段如下图所示：（这里使用reduce函数来进行combine操作，见上面源码）



2、Partitioner阶段

Shuffle一开始就是map阶段做输出操作，一般mapreduce计算的都是海量数据，map输出时候不可能把所有文件都放到内存操作，中间数据是存储在磁盘上的。map写入磁盘的过程十分的复杂，更何况map输出时候要对结果进行排序，内存开销是很大的。map在做输出时候会在内存里开启一个环形内存缓冲区，这个缓冲区专门用来输出的，同时map还会为输出操作启动一个守护线程，如果缓冲区的内存达到了阀值的80%时候，这个守护线程就会把内容写到磁盘上，这个过程叫spill。等map输出全部做完后，map会合并这些输出文件。这个过程里有一个Partitioner操作，Partitioner操作和map阶段的输入分片（Input split）很像。一个Partitioner对应一个reduce作业，如果我们mapreduce操作只有一个reduce，那么Partitioner就只有一个，如果我们有多个reduce，那么Partitioner对应的就会有多个。Partitioner因此就是reduce的输入分片，这个程序员可以编程控制，主要是根据实际key和value的值，根据实际业务类型或者为了更好的reduce负载均衡要求进行，这是提高reduce效率的一个关键所在。到了reduce阶段就是合并map输出文件了，Partitioner会找到对应的map输出文件，然后进行复制操作，复制操作时reduce会开启几个复制线程，这些线程默认个数是5个，程序员也可以在配置文件更改复制线程的个数。这个复制过程和map写入磁盘过程类似，也有阀值和内存大小，阀值一样可以在配置文件里配置，而内存大小是直接使用reduce的tasktracker的内存大小，复制时候reduce还会进行排序操作和合并文件操作，这些操作完了就会进行reduce计算了。

Partitioner默认实现是采用哈希函数：hash(key) mod R，其中

R是Reduce Task数目，该过程允许用户自定义。

对于本程序而言，只有一个Reduce Task，所以两个map的输出会归并到该reduce下，如下图所示：





从图中可以看出，正真reduce方法接收到数据时，value的形式是

Iterable<IntWritable> values

的形式，从上述源码中也可以看出来。

reduce阶段

Reduce方法需要继承org.apache.hadoop.mapreduce包中Reducer类，并重写其reduce方法。Map过程输出

<key,values>

中key为单个单词，而values是对应单词的计数值所组成的列表。Map的输出就是Reduce的输入，所以reduce方法只要遍历values并求和，即可得到某个单词的总次数。其源码如下：

public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
System.out.println("ReduceIn key:"+key.toString()+" value:"+values.toString());
for (IntWritable val : values) {
System.out.println("vlaue:" + val.toString());
sum += val.get();
}
result.set(sum);
System.out.println("ReduceOut key:" + key.toString() + " valueSum:" + result.toString());
context.write(key, result);
}
}

reduce过程如下图：

3、从整体上来看MapReduce

这里给出MapReduce1.0的整体处理流程，对于MapReduce2.0，只是把jobTracker的任务分为两部分，其他逻辑基本一致。参考YARN的运行流程即可。

三、程序提交集群测试

在本地eclipse进行开发完成后，可以把程序打成jar包，放到集群上去运行。基本过程如下：

1、项目打包

右键项目，选择“export”选项，选择java下的jar选项。如图：



点击下一步，选择jar包存放位置和名字，其他默认即可。点击finish完成。

2、将jar包上传至linux集群，此过程略。

3、进入jar包所在目录，在该目录输入如下命令启动任务：

hadoop jar wordcount.jar com.kang.WordCount /root/input /root/output

注意：我的linux上已经把hadoop的bin目录加入了环境变量的Path中，所以可以在任何目录下直接运行hadoop命令。

命令中的

wordcount.jar

是我们打包上传的jar包名称，

com.kang.WordCount

是main方法所在的类的全类名（包含包名），

/root/input

是输入目录，

/root/output

是输出目录。需要注意的是输出目录不能在HDFS上已存在，否则hadoop运行时会报错。

事实上对于本次实例来说，不必指定输入输出目录，因为我在源程序中已经把目录指定了（见上面源码）。如果我们查看hadoop的wordcount原版源码，可以看到其main方法如下：

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length < 2) {
System.err.println("Usage: wordcount <in> [<in>...] <out>");
System.exit(2);
}
Job job = new Job(conf, "word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
for (int i = 0; i < otherArgs.length - 1; ++i) {
FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
}
FileOutputFormat.setOutputPath(job,
new Path(otherArgs[otherArgs.length - 1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}

可以看到它是通过命令行参数指定的输入输出目录。实际开发中我们可以灵活使用。

运行结果如下图：









4、查看结果：

通过命令：

hadoop fs -ls /root/output

可查看输出目录文件列表。

通过命令：

hadoop fs -cat /root/output/part-r-00000

查看结果。

四、常见MapReduce应用场景

简单的数据统计，比如网站pv、uv统计

搜索引擎建索引

海量数据查找

复杂数据分析算法实现，例如

聚类算法

分类算法

推荐算法

图算法