Hadoop大数据教程:Pipes接口怎么实现
Hadoop大数据教程:Pipes接口怎么实现,Hadoop管道与流不同,流使用标准输入和输出让用户的Map和Reduce节点之间相互交流。而管道使用socket作为tasktracker与用户MapReduce进程之间的通道。下面我们同样使用Hadoop发行版中自带单词统计的例子,这个Map和Reduce函数是使用C/C++编写的,然后使用管道机制在Hadoop集群中运行用户的作业,在使用Hadoop Pipes时需要在头文件中包含hadoop/Pipes.hh、hadoop/TemplateFactory.hh、hadoop/StringUtils.hh这3个头文件。
需要注意的是Hadoop版本自带的Pipes例子中有三个实现,对应的源码文件分别为:wordcount-simple.cc、wordcount-part.cc及wordcount-nopipe.cc。下面对这三个文件进行说明。
wordcount-simple.cc,是最基本的Pipes编程,Mapper和Reducer使用C++编写,
RecordReader、Partitioner和RecordWriter均采用Hadoop内置的Java实现,其中,RecordReader为LineRecordReader(位于InputTextInputFormat中,按行读取数据,行所在的偏移量为key,行中的字符串为value),Partitioner为PipesPartitioner,RecordWriter为LineRecordWriter(位于InputTextOutputFormat中,输出格式为“key\tvalue\n”)。
wordcount-part.cc,Mapper、Partitioner和Reducer组件采用C++语言编写,其他组件采用Hadoop的java内置实现。
wordcount-nopipe.cc,RecordReader、Mapper、Reducer和RecordWriter均采用C++编写实现。
接下来介绍最简单的wordcount-simple.cc的Pipes的C++实现。
(1)Map实现
要实现WordCountMap就需要继承HadoopPipes::Mapper类,并实现Map函数,其代码如下:
class WordCountMap: public HadoopPipes: :Mapper {
public:
HadoopPipes: :TaskContext: :Counter* inputWords;
void map(HadoopPipes: :MapContext& context) {
std: :vector words =
HadoopUtils: :splitString(context.getInputValue(), " ");
for(unsigned int i=0; i < words.size(); ++i) {
context.emit(words[i], "1");
}
context.incrementCounter(inputWords, words.size());
}
};
从上述代码中很容易可以看出,Pipes接口是比较类似Java API接口的,通过MapContext类中的方法getInputValue( )获取value,就是每一行的值,处理之后通过MapContext类中的emit方法输出,这个输出也就是下一阶段Reduce的输入。
(2)Redcue实现
要实现WordCountReduce类,就需要继承HadoopPipes::Reducer类,并实现Reduce函数,代码如下:
class WordCountReduce: public HadoopPipes: :Reducer {
public:
HadoopPipes: :TaskContext: :Counter* outputWords;
void reduce(HadoopPipes: :ReduceContext& context) {
int sum = 0;
while (context.nextValue()) {
sum += HadoopUtils: :toInt(context.getInputValue());
}
context.emit(context.getInputKey(), HadoopUtils toString(sum));
context.incrementCounter(outputWords, 1);
}
};
从WordCountReduce类的实现中可以看出,Reduce通过ReduceContext类中的方法nextValue( )来获取当前key的下一个值value,然后对其值相加,最后通过emit方法提交键值对作为Redcue的最终输出。
(3)main函数
Map和Reduce实现之后,还需要一个main函数来驱动,需要调用HadoopPipes::runTask函数来驱动,代码如下:
int main(int argc, char *argv[]) {
return HadoopPipes: :runTask(HadoopPipes: :TemplateFactory
WordCountReduce>());
}
从main函数中可以看到,HadoopPipes::runTask函数有两个参数,分别就是Map和Reduce类。
-
标签错误:<!-- #Label#
labelId=20160707140604
moduleId=1
classId=12231768634
orderby=2
fields=url,title,u_info
attribute=
datatypeId=22192428132
recordCount=3
pageSize=
<htmlTemplate><dt><img src="/images/index_26${index}.jpg" width="100" height="62" /><a href="$url" title="${title}">${title}</a><span>${api.left(u_info,60)}</span></dt></htmlTemplate>
-->
- 我要参加技术沙龙