如何在Hadoop上编写MapReduce程序

用户配置并将一个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接口

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

其中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中必须实现的函数是

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提供的一些工具。

MichaelG.Noll在他的Blog中提到如何在Hadoop中用Python编写MapReduce程序，韩国的gogamza在其Bolg中也提到如何用C编写MapReduce程序（我稍微修改了一下原程序,因为他的Map对单词切分使用tab键）。我合并他们两人的文章，也让国内的Hadoop用户能够使用别的语言来编写MapReduce程序。首先您得配好您的Hadoop集群，这方面的介绍网上比较多，这儿给个链接（Hadoop学习笔记二安装部署）。HadoopStreaming帮助我们用非Java的编程语言使用MapReduce，Streaming用STDIN(标准输入)和STDOUT(标准输出)来和我们编写的Map和Reduce进行数据的交换数据。任何能够使用STDIN和STDOUT都可以用来编写MapReduce程序，比如我们用Python的sys.stdin和sys.stdout，或者是C中的stdin和stdout。我们还是使用Hadoop的例子WordCount来做示范如何编写MapReduce，在WordCount的例子中我们要解决计算在一批文档中每一个单词的出现频率。首先我们在Map程序中会接受到这批文档每一行的数据，然后我们编写的Map程序把这一行按空格切开成一个数组。并对这个数组遍历按"1"用标准的输出输出来，代表这个单词出现了一次。在Reduce中我们来统计单词的出现频率。PythonCodeMap:mapper.py#!/usr/bin/envpythonimportsys#mapswordstotheircountsword2count={}#inputcomesfromSTDIN(standardinput)forlineinsys.stdin:#removeleadingandtrailingwhitespaceline=line.strip()#splitthelineintowordswhileremovinganyemptystringswords=filter(lambdaword:word,line.split())#increasecountersforwordinwords:#writetheresultstoSTDOUT(standardoutput)#whatweoutputherewillbetheinputforthe#Reducestep,i.e.theinputforreducer.py##tab-delimitedthetrivialwordcountis1print'%s\t%s'%(word,1)Reduce:reducer.py#!/usr/bin/envpythonfromoperatorimportitemgetterimportsys#mapswordstotheircountsword2count={}#inputcomesfromSTDINforlineinsys.stdin:#removeleadingandtrailingwhitespaceline=line.strip()#parsetheinputwegotfrommapper.pyword,count=line.split()#convertcount(currentlyastring)tointtry:count=int(count)word2count[word]=word2count.get(word,0)+countexceptValueError:#countwasnotanumber,sosilently#ignore/discardthislinepass#sortthewordslexigraphically##thisstepisNOTrequired,wejustdoitsothatour#finaloutputwilllookmoreliketheofficialHadoop#wordcountexamplessorted_word2count=sorted(word2count.items(),key=itemgetter(0))#writetheresultstoSTDOUT(standardoutput)forword,countinsorted_word2count:print'%s\t%s'%(word,count)CCodeMap:Mapper.c#include#include#include#include#defineBUF_SIZE2048#defineDELIM"\n"intmain(intargc,char*argv[]){charbuffer[BUF_SIZE]while(fgets(buffer,BUF_SIZE-1,stdin)){intlen=strlen(buffer)if(buffer[len-1]=='\n')buffer[len-1]=0char*querys=index(buffer,'')char*query=NULLif(querys==NULL)continuequerys+=1/*nottoinclude'\t'*/query=strtok(buffer,"")while(query){printf("%s\t1\n",query)query=strtok(NULL,"")}}return0}h>h>h>h>Reduce:Reducer.c#include#include#include#include#defineBUFFER_SIZE1024#defineDELIM"\t"intmain(intargc,char*argv[]){charstrLastKey[BUFFER_SIZE]charstrLine[BUFFER_SIZE]intcount=0*strLastKey='\0'*strLine='\0'while(fgets(strLine,BUFFER_SIZE-1,stdin)){char*strCurrKey=NULLchar*strCurrNum=NULLstrCurrKey=strtok(strLine,DELIM)strCurrNum=strtok(NULL,DELIM)/*necessarytocheckerrorbut.*/if(strLastKey[0]=='\0'){strcpy(strLastKey,strCurrKey)}if(strcmp(strCurrKey,strLastKey)){printf("%s\t%d\n",strLastKey,count)count=atoi(strCurrNum)}else{count+=atoi(strCurrNum)}strcpy(strLastKey,strCurrKey)}printf("%s\t%d\n",strLastKey,count)/*flushthecount*/return0}h>h>h>h>首先我们调试一下源码：chmod+xmapper.pychmod+xreducer.pyecho"foofooquuxlabsfoobarquux"|./mapper.py|./reducer.pybar1foo3labs1quux2g++Mapper.c-oMapperg++Reducer.c-oReducerchmod+xMapperchmod+xReducerecho"foofooquuxlabsfoobarquux"|./Mapper|./Reducerbar1foo2labs1quux1foo1quux1你可能看到C的输出和Python的不一样,因为Python是把他放在词典里了.我们在Hadoop时,会对这进行排序,然后相同的单词会连续在标准输出中输出.在Hadoop中运行程序首先我们要下载我们的测试文档wget页面中摘下的用php编写的MapReduce程序,供php程序员参考：Map:mapper.php#!/usr/bin/php$word2count=array()//inputcomesfromSTDIN(standardinput)while(($line=fgets(STDIN))!==false){//removeleadingandtrailingwhitespaceandlowercase$line=strtolower(trim($line))//splitthelineintowordswhileremovinganyemptystring$words=preg_split('/\W/',$line,0,PREG_SPLIT_NO_EMPTY)//increasecountersforeach($wordsas$word){$word2count[$word]+=1}}//writetheresultstoSTDOUT(standardoutput)//whatweoutputherewillbetheinputforthe//Reducestep,i.e.theinputforreducer.pyforeach($word2countas$word=>$count){//tab-delimitedecho$word,chr(9),$count,PHP_EOL}?>Reduce:mapper.php#!/usr/bin/php$word2count=array()//inputcomesfromSTDINwhile(($line=fgets(STDIN))!==false){//removeleadingandtrailingwhitespace$line=trim($line)//parsetheinputwegotfrommapper.phplist($word,$count)=explode(chr(9),$line)//convertcount(currentlyastring)toint$count=intval($count)//sumcountsif($count>0)$word2count[$word]+=$count}//sortthewordslexigraphically////thissetisNOTrequired,wejustdoitsothatour//finaloutputwilllookmoreliketheofficialHadoop//wordcountexamplesksort($word2count)//writetheresultstoSTDOUT(standardoutput)foreach($word2countas$word=>$count){echo$word,chr(9),$count,PHP_EOL}?>作者：马士华发表于：2008-03-05

hadoop有个C++的接口，没有c的。需要自己编译链接库文件。你可以在网上找找，有个wordcount的例子。但是具体讲哪些类的好像没见过。

c语言或者其他语言的程序可以使用streaming方式来运行。

streaming的执行方法如下：

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar [options]

Options:

-input<path>任务的输入路径

-output <path>任务的输出路径，需要是一个还不存在的目录。否则任务不能执行。

-mapper <cmd|JavaClassName> mapper的程序（可以是java的类，也可以是编译好的其他语言的程序）

-combiner <JavaClassName>Combiner的程序

-reducer <cmd|JavaClassName> reducer的程序

-file <file>程序运行所需要的文件或目录，会在执行任务前拷贝到所有节点机器上

-inputformat TextInputFormat(default)|SequenceFileAsTextInputFormat|JavaClassName Optional.输入文件的格式一般保持默认（默认可以不写这个参数）即可。

-outputformat TextOutputFormat(default)|JavaClassName Optional.输出文件的格式

-partitioner JavaClassName 这个是决定根据key值制定处理数据用哪个reducer的那个类。

-numReduceTasks <num> reducer的数目（这只是任务分配的数目，具体能否并行执行还看hadoop配置的最大并发reducer数目）

-inputreader <spec> 这个参数没用过。。。。。

-cmdenv <n>=<v> 运行程序需要的一些环境变量

-mapdebug <path> 远程调试用，不明白是怎么用的。。。

-reducedebug <path> 远程调试用

实际使用中至少需要指定输入、输出目录和mapper程序

例子：$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar -input test -output out -mapper cat

注意：hadoop-streaming.jar默认并不在$HADOOP_HOME中，你可以在$HADOOP_HOME/contrib/streaming/中找到这个jar包，或者直接使用$HADOOP_HOME/contrib/streaming/hadoop-0.20.2-streaming.jar

---