Question

如何在Hadoop上编写MapReduce程序

Answer 1

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

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编程》