中国唯一一家专业只做ARACLE认证和BDA实训的甲骨文金牌合作企业
中国唯一一家专业只做ARACLE认证和BDA实训的甲骨文金牌合作企业
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类。
CUUG 优技培训
CUUG －CHINA UNIX USER GROUP，是国际UNIX组织UNIFORUM的中国代表，是国内悠久的专业UNIX培训机构，被誉为中国UNIX 的摇篮。金牌讲师
实操环境
院校合作
学校新闻
行业新闻
请输入您的手机号
申请试听
