用户登录
用户注册

分享至

mapreduce 程序

  • 作者: 裤裆放不下
  • 来源: 51数据库
  • 2020-09-26
用户配置并将一个Hadoop作业提到Hadoop框架中,Hadoop框架会把这个作业分解成一系列map tasks 和reduce tasks。Hadoop框架负责task分发和执行,结果收集和作业进度监控。

在编写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/ )
关于Pydoop编程,可参见其官方网站:http://sourceforge.net/projects/pydoop/
关于Hadoop pipes编程,可参见《Hadoop Tutorial 2.2 — Running C++ Programs on Hadoop》。
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=,。
8. 参考资料
(1) 书籍 Jason Venner 《Pro Hadoop》
(2) 书籍 Chuck Lam 《Hadoop in Action》
(3) 书籍 Tom White 《Hadoop The Definitive Guide》
(4) Hadoop分布式缓存例子:书籍《Hadoop The Definitive Guide》 第八章 最后一节“Distributed Cache”
(5) Hadoop Pipes例子:源代码中$HAOOOP_HOME/src/examples/pipes路径下。
(6) Hadoop Pipes资料:
http://developer.yahoo.com/hadoop/tutorial/module4.html#pipes
http://wiki.apache.org/hadoop/C%2B%2BWordCount
软件
前端设计
程序设计
Java相关