当前位置: Oracle DBA培训网-优技培训 >> Hadoop大数据 > Hadoop培训教程 >

Hadoop大数据教程:Pipes接口怎么实现

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类。

技术沙龙MORE+

标签错误:<!-- #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> -->
我要参加技术沙龙