如何在Hadoop上编写MapReduce程序

1. 概述

1970年，IBM的研究员E.F.Codd博士在刊物《Communication of the ACM》上发表了一篇名为“A Relational Model of Data for Large Shared Data Banks”的论文，提出了关系模型的概念，标志着关系数据库的诞生，随后几十年，关系数据库及其结构化查询语言SQL成为程序员必须掌握的基本技能之一。

2005年4月，Jeffrey Dean和Sanjay Ghemawat在国际会议OSDI上发表“MapReduce: Simplified Data Processing on Large Cluster”，标志着google的大规模数据处理系统MapReduce公开。受这篇论文的启发，当年秋天，Hadoop 由 Apache Software Foundation 公司作为 Lucene 的子项目 Nutch 的一部分正式被引入，2006 年 3 月份，MapReduce 和 Nutch Distributed File System (NDFS) 分别被纳入称为 Hadoop 的项目中。如今，Hadoop已经被超过50%的互联网公司使用，其他很多公司正准备使用Hadoop来处理海量数据，随着Hadoop越来越受欢迎，也许在将来的某段时间，Hadoop会成为程序员必须掌握的技能之一，如果真是这样的话，学会如何在Hadoop上编写MapReduce程序便是学习Hadoop的开始。

本文介绍了在Hadoop上编写MapReduce程序的基本方法，包括MapReduce程序的构成，不同语言开发MapReduce的方法等。

2. Hadoop 作业构成

2.1 Hadoop作业执行流程

用户配置并将一个Hadoop作业提到Hadoop框架中，Hadoop框架会把这个作业分解成一系列map tasks 和reduce tasks。Hadoop框架负责task分发和执行，结果收集和作业进度监控。

下图给出了一个作业从开始执行到结束所经历的阶段和每个阶段被谁控制（用户 or Hadoop框架）。

下图详细给出了用户编写MapRedue作业时需要进行那些工作以及Hadoop框架自动完成的工作：

在编写MapReduce程序时，用户分别通过InputFormat和OutputFormat指定输入和输出格式，并定义Mapper和Reducer指定map阶段和reduce阶段的要做的工作。在Mapper或者Reducer中，用户只需指定一对key/value的处理逻辑，Hadoop框架会自动顺序迭代解析所有key/value，并将每对key/value交给Mapper或者Reducer处理。表面上看来，Hadoop限定数据格式必须为key/value形式，过于简单，很难解决复杂问题，实际上，可以通过组合的方法使key或者value（比如在key或者value中保存多个字段，每个字段用分隔符分开，或者value是个序列化后的对象，在Mapper中使用时，将其反序列化等）保存多重信息，以解决输入格式较复杂的应用。

2.2 用户的工作

用户编写MapReduce需要实现的类或者方法有：

（1） InputFormat接口

用户需要实现该接口以指定输入文件的内容格式。该接口有两个方法

public interface InputFormat<K, V> {
 
     InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
 
     RecordReader<K, V> getRecordReader(InputSplit split,
 
     JobConf job,
 
     Reporter reporter) throws IOException;
 
}
复制代码

其中getSplits函数将所有输入数据分成numSplits个split，每个split交给一个map task处理。getRecordReader函数提供一个用户解析split的迭代器对象，它将split中的每个record解析成key/value对。

Hadoop本身提供了一些InputFormat：

（2）Mapper接口

用户需继承Mapper接口实现自己的Mapper，Mapper中必须实现的函数是

void map(K1 key,
 
    V1 value,
 
    OutputCollector<K2,V2> output,
 
    Reporter reporter
 
) throws IOException
复制代码

其中，<K1 V1>是通过Inputformat中的RecordReader对象解析处理 的，OutputCollector获取map()的输出结果，Reporter保存了当前task处理进度。

Hadoop本身提供了一些Mapper供用户使用：

（3）Partitioner接口

用户需继承该接口实现自己的Partitioner以指定map task产生的key/value对交给哪个reduce task处理，好的Partitioner能让每个reduce task处理的数据相近，从而达到负载均衡。Partitioner中需实现的函数是

getPartition(  K2   key, V2 value, int numPartitions)

该函数返回<K2 V2>对应的reduce task ID。

用户如果不提供Partitioner，Hadoop会使用默认的（实际上是个hash函数）。

（4）Combiner

Combiner使得map task与reduce task之间的数据传输量大大减小，可明显提高性能。大多数情况下，Combiner与Reducer相同。

（5）Reducer接口

用户需继承Reducer接口实现自己的Reducer，Reducer中必须实现的函数是

void reduce(K2 key,
 
     Iterator<V2> values,
 
     OutputCollector<K3,V3> output,
 
     Reporter reporter
 
) throws IOException
复制代码

Hadoop本身提供了一些Reducer供用户使用：

（6）OutputFormat

用户通过OutputFormat指定输出文件的内容格式，不过它没有split。每个reduce task将其数据写入自己的文件，文件名为part-nnnnn，其中nnnnn为reduce task的ID。

Hadoop本身提供了几个OutputFormat:

3. 分布式缓存

Haoop中自带了一个分布式缓存，即DistributedCache对象，方便map task之间或者reduce task之间共享一些信息，比如某些实际应用中，所有map task要读取同一个配置文件或者字典，则可将该配置文件或者字典放到分布式缓存中。

4. 多语言编写MapReduce作业

Hadoop采用java编写，因而Hadoop天生支持java语言编写作业，但在实际应用中，有时候，因要用到非java的第三方库或者其他原因，要采用C/C++或者其他语言编写MapReduce作业，这时候可能要用到Hadoop提供的一些工具。

如果你要用C/C++编写MpaReduce作业，可使用的工具有Hadoop Streaming或者Hadoop Pipes。

如果你要用Python编写MapReduce作业，可以使用Hadoop Streaming或者Pydoop。

如果你要使用其他语言，如shell，php，ruby等，可使用Hadoop Streaming。

关于Hadoop Streaming编程，可参这篇博文：《Hadoop Streaming编程》（http://dongxicheng.org/mapreduce/hadoop-streaming-programming/ ）

5. 编程方式比较

（1）java。 Hadoop支持的最好最全面的语言，而且提供了很多工具方便程序员开发。

（2）Hadoop Streaming。 它最大的优点是支持多种语言，但效率较低，reduce task需等到map 阶段完成后才能启动；它不支持用户自定义InputFormat，如果用户想指定输入文件格式，可使用java语言编写或者在命令行中指定分隔符；它采用标准输入输出让C/C++与java通信，因而只支持text数据格式。

（3）Hadoop Pipes。 专门为C/C++语言设计，由于其采用了socket方式让C/C++与java通信，因而其效率较低（其优势在于，但作业需要大量，速度很快）。它支持用户（用C/C++）编写RecordReader。

（4）Pydoop。它是专门方便python程序员编写MapReduce作业设计的，其底层使用了Hadoop Streaming接口和libhdfs库。

6. 总结

Hadoop使得分布式程序的编写变得异常简单，很多情况下，用户只需写map()和reduce()两个函数即可（InputFormat，Outputformat可用系统缺省的）。正是由于Hadoop编程的简单性，越来越多的公司或者研究单位开始使用Hadoop。

7. 注意事项

（1） Hadoop默认的InputFormat是TextInputFormat，它将文件中的每一行作为value，该行的偏移量为key。

（2）如果你的输入是文本文件，且每一行包括key，value，则可使用Hadoop中自带的KeyValueTextInputFormat，它默认的每行是一个key/value对，且key与value的分割如为TAB（’\t‘），如果想修改key/value之间的分隔符，如将分割符改为“，”,可使用conf.set(“key.value.separator.in.input.line”,”,”);或者-D key.value.separator.in.input.line=，。

懂西城

cool，这个好，我来测试下。
另外，楼主用什么方式粘贴code的啊，look beautiful。

如上图蓝色圈所示，高级编辑中有“代码”功能。

弱问：哪位能告诉我上文 mapper.c的第十六行，那个
char *querys  = index(buffer, ' ');
            char *query = NULL;
index函数功能是什么

回复 4# cccomeon
下面是摘自它的Linux MAN手册，应当是比较清楚了：
NAME
       index, rindex - locate character in string
SYNOPSIS
       #include
       char *index(const char *s, int c);
       char *rindex(const char *s, int c);
RETURN VALUE
       The  index()  and  rindex()  functions  return  a pointer to the matched character or NULL if the character is not
       found.
index和rindex一个从左扫描，另一个从右（right）扫描。
示例：
#include
#include
int main() {
    char str[] = "abc
[color=]d
efghijklmn";
    char *p = index(str, '
[color=]d
');
    printf("%s\n", p);
    printf("%d\n", p-str);
    return 0;
}
运行结果：

[color=]d
efghijklmn
3

再问一下，mapper.c 第26行
printf("%s\t1\n", query);
\t1 是什么意思呢吗？

你要把\t和1拆开来看，\t是tab，就是空4个格，1就是个数字1。
所以上面的意思是先输出query，然后空个tab的位置，再输出1.
比如： string    1

回复 1# eyjian
关于文中提到的
用Hadoop Streaming运行MapReduce会比较用Java的代码要慢,因为有两方面的原因:

使用 Java API >> C Streaming >> Perl Streaming 这样的一个流程运行会阻塞IO.

不像Java在运行Map后输出结果有一定数量的结果集就启动Reduce的程序,用Streaming要等到所有的Map都运行完毕后才启动Reduce   为什么用Streaming要等到所有的Map都运行完毕后才启动Reduce而Java在运行Map后输出结果有一定数量的结果集就启动Reduce的程序，能说的详细点吗吗？一直没明白为什么Java运行的Map输出结果有一定数量的结果集就启动Reduce的程序
我看源代码中StreamJob只是把命令都封装到了PipeMapRed中进行执行的
如果我写一个MAP，在其中用Process执行外部程序命令并处理输出结果，这跟Streaming模式执行外部命令相比有什么性能差异吗吗？

楼主好，我运行的时候出现了：
[root@node161 hadoop]# bin/hadoop jar contrib/streaming/hadoop-0.20.2-streaming.jar -mapper /root/mapper.py -reducer /root/reducer.py -input input/* -output output
packageJobJar: [/home/tmp/hadoop-unjar1693441522665847631/] [] /tmp/streamjob3333054264377001703.jar tmpDir=null
10/04/13 19:32:01 INFO mapred.FileInputFormat: Total input paths to process : 2
10/04/13 19:32:02 INFO streaming.StreamJob: getLocalDirs(): [/home/tmp//mapred/local]
10/04/13 19:32:02 INFO streaming.StreamJob: Running job: job_201004122206_0015
10/04/13 19:32:02 INFO streaming.StreamJob: To kill this job, run:
10/04/13 19:32:02 INFO streaming.StreamJob: /opt/hadoop/bin/../bin/hadoop job  -Dmapred.job.tracker=node161:9001 -kill job_201004122206_0015
10/04/13 19:32:02 INFO streaming.StreamJob: Tracking URL: http://node161:50030/jobdetails.jsp?jobid=job_201004122206_0015
10/04/13 19:32:03 INFO streaming.StreamJob:  map 0%  reduce 0%
10/04/13 19:32:34 INFO streaming.StreamJob:  map 100%  reduce 100%
10/04/13 19:32:34 INFO streaming.StreamJob: To kill this job, run:
10/04/13 19:32:34 INFO streaming.StreamJob: /opt/hadoop/bin/../bin/hadoop job  -Dmapred.job.tracker=node161:9001 -kill job_201004122206_0015
10/04/13 19:32:34 INFO streaming.StreamJob: Tracking URL: http://node161:50030/jobdetails.jsp?jobid=job_201004122206_0015
10/04/13 19:32:34 ERROR streaming.StreamJob: Job not Successful!
10/04/13 19:32:34 INFO streaming.StreamJob: killJob...
Streaming Job Failed!
请问这是怎么回事儿吗？谢谢！

哈哈，搞定了！原因是没有将mapper.py和reducer.py拷贝到各个子节点去~