mapreduce采用”分而治之”的思想，把对大规模数据集的操作，分发给一个主节点管理下的各个分节点共同完成，然后通过整合各个节点的中间结果，得到最终结果。简单地说，mapreduce就是”任务的分解与结果的汇总”。
在hadoop中，用于执行mapreduce任务的机器角色有两个：一个是jobtracker；另一个是tasktracker，jobtracker是用于调度工作的，tasktracker是用于执行工作的。一个hadoop集群中只有一台jobtracker。
在分布式计算中，mapreduce框架负责处理了并行编程中分布式存储、工作调度、负载均衡、容错均衡、容错处理以及网络通信等复杂问题，把处理过程高度抽象为两个函数：map和reduce，map负责把任务分解成多个任务，reduce负责把分解后多任务处理的结果汇总起来。
需要注意的是，用mapreduce来处理的数据集（或任务）必须具备这样的特点：待处理的数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全并行地进行处理。
在hadoop中，每个mapreduce任务都被初始化为一个job，每个job又可以分为两种阶段：map阶段和reduce阶段。这两个阶段分别用两个函数表示，即map函数和reduce函数。map函数接收一个<key,value>形式的输入，然后同样产生一个<key,value>形式的中间输出，hadoop函数接收一个如<key,(list of values)>形式的输入，然后对这个value集合进行处理，每个reduce产生0或1个输出，reduce的输出也是<key,value>形式的。
mapreduce处理大数据集的过程
单词计数是最简单也是最能体现mapreduce思想的程序之一，可以称为mapreduce版”hello world”，该程序的完整代码可以在hadoop安装包的”src/examples”目录下找到。单词计数主要完成功能是：统计一系列文本文件中每个单词出现的次数，如下图所示。
现在以”hadoop”普通用户登录”master.hadoop”服务器。
1）创建本地示例文件
首先在”/home/hadoop”目录下创建文件夹”file”。
接着创建两个文本文件file1.txt和file2.txt，使file1.txt内容为”hello world”，而file2.txt的内容为”hello hadoop”。
2）在hdfs上创建输入文件夹
3）上传本地file中文件到集群的input目录下
1）在集群上运行wordcount程序
备注： 以input作为输入目录，output目录作为输出目录。
已经编译好的wordcount的jar在”/usr/hadoop”下面，就是”hadoop-examples-1.0.0.jar”，所以在下面执行命令时记得把路径写全了，不然会提示找不到该jar包。
2）mapreduce执行过程显示信息
hadoop命令会启动一个jvm来运行这个mapreduce程序，并自动获得hadoop的配置，同时把类的路径（及其依赖关系）加入到hadoop的库中。以上就是hadoop job的运行记录，从这里可以看到，这个job被赋予了一个id号：job_201202292213_0002，而且得知输入文件有两个（total input paths to process : 2），同时还可以了解map的输入输出记录（record数及字节数），以及reduce输入输出记录。比如说，在本例中，map的task数量是2个，reduce的task数量是一个。map的输入record数是2个，输出record数是4个等信息。
1）查看hdfs上output目录内容
从上图中知道生成了三个文件，我们的结果在” part-r-00000 “中。
2）查看结果输出文件内容
hadoop提供了如下内容的数据类型，这些数据类型都实现了writablecomparable接口，以便用这些类型定义的数据可以被序列化进行网络传输和文件存储，以及进行大小比较。
booleanwritable：标准布尔型数值
bytewritable：单字节数值
doublewritable：双字节数
floatwritable：浮点数
intwritable：整型数
longwritable：长整型数
text：使用utf8格式存储的文本
nullwritable：当<key,value>中的key或value为空时使用
1）源代码程序
package org.apache.hadoop.examples; import java.io.ioexception; import java.util.iterator; import java.util.stringtokenizer; import org.apache.hadoop.fs.path; import org.apache.hadoop.io.intwritable; import org.apache.hadoop.io.longwritable; import org.apache.hadoop.io.text; import org.apache.hadoop.mapred.fileinputformat; import org.apache.hadoop.mapred.fileoutputformat; import org.apache.hadoop.mapred.jobclient; import org.apache.hadoop.mapred.jobconf; import org.apache.hadoop.mapred.mapreducebase; import org.apache.hadoop.mapred.mapper; import org.apache.hadoop.mapred.outputcollector; import org.apache.hadoop.mapred.reducer; import org.apache.hadoop.mapred.reporter; import org.apache.hadoop.mapred.textinputformat; import org.apache.hadoop.mapred.textoutputformat; public class wordcount { public static class map extends mapreducebase implements mapper<longwritable, text, text, intwritable> { private final static intwritable one = new intwritable(1); private text word = new text(); public void map(longwritable key, text value, outputcollector<text, intwritable> output, reporter reporter) throws ioexception { string line = value.tostring(); stringtokenizer tokenizer = new stringtokenizer(line); while (tokenizer.hasmoretokens()) { word.set(tokenizer.nexttoken()); output.collect(word, one); } } } public static class reduce extends mapreducebase implements reducer<text, intwritable, text, intwritable> { public void reduce(text key, iterator<intwritable> values, int sum = 0; while (values.hasnext()) { sum += values.next().get(); output.collect(key, new intwritable(sum)); public static void main(string[] args) throws exception { jobconf conf = new jobconf(wordcount.class); conf.setjobname(“wordcount”); conf.setoutputkeyclass(text.class); conf.setoutputvalueclass(intwritable.class); conf.setmapperclass(map.class); conf.setcombinerclass(reduce.class); conf.setreducerclass(reduce.class); conf.setinputformat(textinputformat.class); conf.setoutputformat(textoutputformat.class); fileinputformat.setinputpaths(conf, new path(args[0])); fileoutputformat.setoutputpath(conf, new path(args[1])); jobclient.runjob(conf); } }
3）主方法 main 分析
public static void main(string[] args) throws exception { jobconf conf = new jobconf(wordcount.class); conf.setjobname(“wordcount”); conf.setoutputkeyclass(text.class); conf.setoutputvalueclass(intwritable.class); conf.setmapperclass(map.class); conf.setcombinerclass(reduce.class); conf.setreducerclass(reduce.class); conf.setinputformat(textinputformat.class); conf.setoutputformat(textoutputformat.class); fileinputformat.setinputpaths(conf, new path(args[0])); fileoutputformat.setoutputpath(conf, new path(args[1])); jobclient.runjob(conf);
首先讲解一下 job 的 初始化过程 。 main 函数调用 jobconf 类来对 mapreduce job进行初始化，然后调用 setjobname() 方法命名这个 job 。对job进行合理的命名有助于 更快 地找到job，以便在jobtracker和tasktracker的页面中对其进行 监视 。
jobconf conf = new jobconf(wordcount. class ); conf.setjobname(“wordcount” );
接着设置job输出结果<key,value>的中key和value数据类型，因为结果是<单词,个数>，所以key设置为”text”类型，相当于java中string类型。value设置为”intwritable”，相当于java中的int类型。
conf.setoutputkeyclass(text.class ); conf.setoutputvalueclass(intwritable.class );
然后设置job处理的map（拆分）、combiner（中间结果合并）以及reduce（合并）的相关处理类。这里用reduce类来进行map产生的中间结果合并，避免给网络数据传输产生压力。
conf.setmapperclass(map.class ); conf.setcombinerclass(reduce.class ); conf.setreducerclass(reduce.class );
接着就是调用setinputpath()和setoutputpath()设置输入输出路径。
conf.setinputformat(textinputformat.class ); conf.setoutputformat(textoutputformat.class );
（1）inputformat和inputsplit
inputsplit是hadoop定义的用来 传送 给每个 单独 的 map 的 数据 ，inputsplit 存储 的并 非 数据本身 ， 而是一个 分片长度 和一个 记录数据位置 的 数组 。 生成inputsplit的方法 可以通过 inputformat() 来 设置 。
当数据传送给 map 时，map会将输入 分片 传送到 inputformat ，inputformat则 调用方法 getrecordreader() 生成 recordreader ， recordreader再通过 creatkey() 、creatvalue() 方法 创建 可供map处理的 <key,value> 对。简而言之，inputformat()方法是用来生成可供map处理的<key,value>对的。
hadoop预定义了多种方法将不同类型的输入数据转化为map能够处理的<key,value>对，它们都继承自inputformat，分别是：
inputformat | |—baileyborweinplouffe.bbpinputformat |—composableinputformat |—compositeinputformat |—dbinputformat |—distsum.machine.abstractinputformat |—fileinputformat |—combinefileinputformat |—keyvaluetextinputformat |—nlineinputformat |—sequencefileinputformat |—terainputformat |—textinputformat
其中 textinputformat 是hadoop 默认 的输入方法，在textinputformat中，每个文件（或其一部分）都会单独地作为map的输入，而这个是继承自fileinputformat的。之后，每行数据都会生成一条记录，每条记录则表示成<key,value>形式：
key值是每个数据的记录在 数据分片 中 字节偏移量 ，数据类型是 longwritable ；
value值是每行的内容，数据类型是 text 。
（2）outputformat
每一种 输 入 格式 都有一种 输 出 格式 与其对应。默认的输出格式是textoutputformat ，这种输出方式与输入类似，会将每条记录以一行的形式存入文本文件。不过，它的 键和值 可以是 任意形式 的，因为程序 内容 会调用 tostring() 方法将键和值转换为 string 类型再输出。
3）map类中map方法分析
public static class map extends mapreducebase implements mapper<longwritable, text, text, intwritable> { private final static intwritable one = new intwritable(1); private text word = new text(); public void map(longwritable key, text value, outputcollector<text, intwritable> output, reporter reporter) throws ioexception { string line = value.tostring(); stringtokenizer tokenizer = new stringtokenizer(line); while (tokenizer.hasmoretokens()) { word.set(tokenizer.nexttoken()); output.collect(word, one);
map类 继承自 mapreducebase ，并且它实现了 mapper接口 ，此接口是一个 规范类型 ，它有4种形式的参数，分别用来指定map的 输入 key值类型、 输入 value值类型、 输出 key值类型和 输出 value值类型。在本例中，因为使用的是textinputformat，它的输出key值是longwritable类型，输出value值是text类型，所以map的输入类型为<longwritable,text>。在本例中需要输出<word,1>这样的形式，因此输出的key值类型是text，输出的value值类型是intwritable。
实现此接口类还需要实现map方法，map方法会具体负责对输入进行操作，在本例中，map方法对输入的行以空格为单位进行切分，然后使用 outputcollect 收集输出的<word,1>。
4）reduce类中reduce方法分析
public static class reduce extends mapreducebase implements reducer<text, intwritable, text, intwritable> { public void reduce(text key, iterator<intwritable> values, int sum = 0; while (values.hasnext()) { sum += values.next().get(); output.collect(key, new intwritable(sum));
reduce类 也是继承自 mapreducebase 的，需要实现reducer接口。reduce类以map的输出作为输入，因此reduce的输入类型是<text，intwritable>。而reduce的输出是 单词 和 它的数目 ，因此，它的输出类型是<text,intwritable>。reduce类也要实现reduce方法，在此方法中，reduce函数将输入的key值作为输出的key值，然后将获得多个value值加起来，作为输出的值。
import java.io.ioexception; 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; import org.apache.hadoop.util.genericoptionsparser; 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()); 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) int sum = 0; for (intwritable val : values) { sum += val.get(); result.set(sum); context.write(key, result); 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> <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); fileinputformat.addinputpath(job, new path(otherargs[0])); fileoutputformat.setoutputpath(job, new path(otherargs[1])); system.exit(job.waitforcompletion(true) ? 0 : 1);
1）map过程
map过程需要继承org.apache.hadoop.mapreduce包中 mapper 类，并 重写 其map方法。通过在map方法中添加两句把key值和value值输出到控制台的代码，可以发现map方法中value值存储的是文本文件中的一行（以回车符为行结束标记），而key值为该行的首字母相对于文本文件的首地址的偏移量。然后stringtokenizer类将每一行拆分成为一个个的单词，并将<word,1>作为map方法的结果输出，其余的工作都交有mapreduce框架 处理。
2）reduce过程
reduce过程需要继承org.apache.hadoop.mapreduce包中 reducer 类，并 重写 其reduce方法。map过程输出<key,values>中key为单个单词，而values是对应单词的计数值所组成的列表，map的输出就是reduce的输入，所以reduce方法只要遍历values并求和，即可得到某个单词的总次数。
3）执行mapreduce任务
在mapreduce中，由job对象负责管理和运行一个计算任务，并通过job的一些方法对任务的参数进行相关的设置。此处设置了使用tokenizermapper完成map过程中的处理和使用intsumreducer完成combine和reduce过程中的处理。还设置了map过程和reduce过程的输出类型：key的类型为text，value的类型为intwritable。任务的输出和输入 路径 则由命令行参数指定，并由fileinputformat和fileoutputformat分别设定。完成相应任务的参数设定后，即可调用 job.waitforcompletion() 方法执行任务。
本节将对wordcount进行更详细的讲解。详细执行步骤如下：
1）将文件拆分成splits，由于测试用的文件较小，所以每个文件为一个split，并将文件按行分割形成<key,value>对，如图4-1所示。这一步由mapreduce框架自动完成，其中偏移量（即key值）包括了回车所占的字符数（windows和linux环境会不同）。
图4-1 分割过程
2）将分割好的<key,value>对交给用户定义的map方法进行处理，生成新的<key,value>对，如图4-2所示。
图4-2 执行map方法
3）得到map方法输出的<key,value>对后，mapper会将它们按照key值进行排序，并执行combine过程，将key至相同value值累加，得到mapper的最终输出结果。如图4-3所示。
图4-3 map端排序及combine过程
4）reducer先对从mapper接收的数据进行排序，再交由用户自定义的reduce方法进行处理，得到新的<key,value>对，并作为wordcount的输出结果，如图4-4所示。
图4-4 reduce端排序及输出结果
hadoop最新版本的mapreduce release 0.20.0的api包括了一个全新的mapreduce java api，有时候也称为上下文对象。
新的api类型上不兼容以前的api，所以，以前的应用程序需要重写才能使新的api发挥其作用 。
新的api和旧的api之间有下面几个明显的区别。
新的api倾向于使用抽象类，而不是接口，因为这更容易扩展。例如，你可以添加一个方法(用默认的实现)到一个抽象类而不需修改类之前的实现方法。在新的api中，mapper和reducer是抽象类。
新的api是在org.apache.hadoop.mapreduce包(和子包)中的。之前版本的api则是放在org.apache.hadoop.mapred中的。
新的api广泛使用context object(上下文对象)，并允许用户代码与mapreduce系统进行通信。例如，mapcontext基本上充当着jobconf的outputcollector和reporter的角色。
新的api同时支持”推”和”拉”式的迭代。在这两个新老api中，键/值记录对被推mapper中，但除此之外，新的api允许把记录从map()方法中拉出，这也适用于reducer。”拉”式的一个有用的例子是分批处理记录，而不是一个接一个。
新的api统一了配置。旧的api有一个特殊的jobconf对象用于作业配置，这是一个对于hadoop通常的configuration对象的扩展。在新的api中，这种区别没有了，所以作业配置通过configuration来完成。作业控制的执行由job类来负责，而不是jobclient，它在新的api中已经荡然无存。