Hadoopnbsp;pipes设计原理

1. 什么是Hadoop pipes&＃xff1f;

Hadoop pipes允许用户使用C&＃43;&＃43;语言进行MapReduce程序设计。它采用的主要方法是将应用逻辑相关的C&＃43;&＃43;代码放在单独的进程中&＃xff0c;然后通过Socket让Java代码与C&＃43;&＃43;代码通信。从很大程度上说&＃xff0c;这种方法类似于Hadoop Streaming&＃xff0c;不同的是&＃xff0c;它采用Writable序列化方法将数据转化为byte&＃xff0c;然后通过socket传给C&＃43;&＃43;进程。

org.apache.hadoop.mapred.pipes.Submitter包中有一个public static方法用于提交作业,该方法将作业封装成一个JobConf对象和一个main方法&＃xff08;接收一个应用程序&＃xff0c;可选的配置文件&＃xff0c;输入目录和输出目录等&＃xff09;&＃xff0c;main方法的CLI(Client Line Interface)如下&＃xff1a;

本文主要介绍了Hadoop pipes的设计原理&＃xff0c;包括设计架构&＃xff0c;设计细节等。

2. Hadoop pipes设计架构

用户通过bin/hadoop pipes将作业提交到org.apache.hadoop.mapred.pipes中的Submmit类&＃xff0c;它首先会进行作业参数配置&＃xff08;调用函数setupPipesJob&＃xff09;&＃xff0c;然后通过JobClient(conf).submitJob(conf)将作业提交到Hadoop集群中。

在函数setupPipesJob中&＃xff0c;Java代码会使用ServerScoket创建服务器对象&＃xff0c;然后通过ProcessBuilder执行C&＃43;&＃43;binary&＃xff0c; C&＃43;&＃43;binary实际上是一个Socket client&＃xff0c;它从Java server中接收key/value数据&＃xff0c;经过处理&＃xff08;map&＃xff0c;partition或者reduce等&＃xff09;后&＃xff0c;返还给Java server&＃xff0c;并由Java Server将数据写到HDFS或者磁盘。

3. Hadoop pipes设计细节

Hadoop pipes允许用户用C&＃43;&＃43;编写五个基本组件&＃xff1a;mapper&＃xff0c;reducer&＃xff0c;partitioner&＃xff0c;combiner&＃xff0c;recordReader&＃xff0c;这五个组件可以是Java编写的&＃xff0c;也可以是C&＃43;&＃43;编写的&＃xff0c;下面分别介绍这几个函数的执行过程。

&＃xff08;1&＃xff09; mapper

Pipes会根据用户的配置定制InputFormat&＃xff0c;如果用户要使用Java的InputFormat&＃xff08;hadoop.pipes.java.recordreader&＃61;true&＃xff09;&＃xff0c;则Hadoop会使用户输入的InputFormat&＃xff08;默认为TextInputFormat&＃xff09;&＃xff1b;如果用户使用C&＃43;&＃43;的InputFormat&＃xff0c;则Pipes Java端的代码会读取每个InputSplit&＃xff0c;并调用downlink.runMap(reporter.getInputSplit(), job.getNumReduceTasks(), isJavaInput);通过socket传输给C&＃43;&＃43;端的runMap(string _inputSplit, int _numReduces, bool pipedInput)函数。

在C&＃43;&＃43;端&＃xff0c;RecordReader会解析整个InputSplit&＃xff0c;获取数据来源&＃xff08;主要是文件路径&＃xff09;和每个key/value对&＃xff0c;并交给map函数处理&＃xff0c;map将每个key/value的处理结果通过emit(const string& key, const string& value)函数返还给Java Server。

&＃xff08;2&＃xff09; paritioner

C&＃43;&＃43;端处理完的结果会通过emit(const string& key, const string& value)函数传给Java Server&＃xff0c;以便将数据写到磁盘上。在emit函数中&＃xff0c;如果用户定义了自己的paritioner&＃xff0c;则Pipes会通过该函数判断当前key/value将给哪个reduce task处理&＃xff0c;并调用partitionedOutput(int reduce, const string& key,const string& value)函数将key/value传递给相应的reduce task。

&＃xff08;3&＃xff09; reducer

reducer的执行过程与mapper基本一致。

4. 总结

Hadoop pipes给C&＃43;&＃43;程序员提供了一个编写MapReduce作业的方案&＃xff0c;它使用socket让Java和C&＃43;&＃43;之间进行通信&＃xff0c;这类似于thrift RPC的原理&＃xff0c;也许Hadoop Pipes用thrift编写会更加简单。

Hadoop pipes使用Java代码从HDFS上读写数据&＃xff0c;并将处理逻辑封装到C&＃43;&＃43;中&＃xff0c;数据会通过socket从Java传输给C&＃43;&＃43;&＃xff0c;这虽然增加了数据传输的代价&＃xff0c;但对于计算密集型的作业&＃xff0c;其性能也许会有改进。

5. 参考资料

http://wiki.apache.org/hadoop/HowToDebugMapReducePrograms

http://cs.smith.edu/dftwiki/index.php/Hadoop_Tutorial_2.2_–_Running_C&＃43;&＃43;_Programs_on_Hadoop

http://www.itberry.com/?p&＃61;42