假设你有上百G的数据，你要统计出这些数据中，含有某些你感兴趣的内容的数据的有多少条，你会怎么做？在硬件条件允许的情况下，用hadoop并行计算是一个不错的选择。

为了使本文得以清晰地说明，我们不妨假设如下的情况：

 

我们有100G的数据，分别保存在5个文件中，它们位于 /data/ 目录下。这5个数据文件的内容均为相同的格式，即，文件的内容大致如下：

ABCDSDFKJDKF kkk 2890876

SDKFJIEORUEW nnn 1231238

LSFKDFSSDRDE bbb 9234999

说明：每一行内容中，首先是一个12字节的字符串，然后是一个3字节的字符串，然后是一个7个数字组成的字符串。字符串之间是用空格分隔的。

我们的问题是：在这100G的数据中，请统计出第二项字符串为“kkk”和“nnn”的数据分别有多少条？

如果用一个非分布式的应用程序来计算这个问题，如果计算机硬件配置不够强劲的话，那么估计得算到天荒地老了。

而用hadoop来并行计算，一切都是那么简单。

下面，我们就来看看，如何用C++开发一个hadoop上的应用程序，来完成我们的任务。

尽管hadoop平台是用Java写的，但是它仍然支持用C++来开发应用程序，这里不讨论优劣对比，只是基于这样一个事实：有些人觉得用C++写更熟悉，所以我们才用C++写。

先说明：本文基于hadoop 0.20.2版本。

文章来源：

（1）首先我们需要知道map-reduce的基本原理，这里不说了。其次我们需要知道，在用C++编写hadoop应用程序时，需要包含三个头文件：

#include "Pipes.hh"

#include "TemplateFactory.hh"

#include "StringUtils.hh"

这三个文件在hadoop安装包的 “c++\Linux-amd64-64\include\” 或 “c++\Linux-i386-32\include\” 子目录下（根据你的操作系统是64位或32位，分别对应不同的目录）。

既然有头文件，就需要有对应的实现文件，或者动态/静态库，这里我用的是静态库 libhadooppipes.a 和 libhadooputils.a 。静态库是在Makefile中指定的，后面再说。这里特别提醒一下大家：如果你的hadoop集群不是只有一台服务器，那么如果你编译时使用了任何动态库的话，在运行的时候就要保证在别的hadoop服务器上也能找到相应的动态库，否则就会在hadoop JobTracker的详细信息中看到找不到动态库的错误提示。

文章来源：

（2）下面来看看程序：

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34

#include"Pipes.hh" #include"TemplateFactory.hh" #include"StringUtils.hh"   class DataCountMap: public HadoopPipes::Mapper { public :    DataCountMap(HadoopPipes::TaskContext&context){}    void map(HadoopPipes::MapContext&context) {      std::vector<std::string>words=HadoopUtils::splitString(context.getInputValue(), " " );    // 这里是分割字符串，如前文所述，每一行数据中的各项是以空格来分割的。分割的结果保存到了一个std::vector中      if ( "kkk" ==words[1]) {        context.emit( "kkk" , "1" );      } else if ( "nnn" ==words[1]) {        context.emit( "nnn" , "1" );      }    } };   class DataCountReduce: public HadoopPipes::Reducer { public :    DataCountReduce(HadoopPipes::TaskContext&context){}    void reduce(HadoopPipes::ReduceContext&context)    {      int sum=0;      while (context.nextValue()) {        sum++;      }      context.emit(context.getInputKey(),HadoopUtils::toString(sum));    } };   int main( int argc, char *argv[]) {    return HadoopPipes::runTask(HadoopPipes::TemplateFactory<DataCountMap, DataCountReduce>()); }

上面的程序挺简单的，只要你知道了map-reduce的基本原理。

一个map类，一个reduce类，一个执行任务的main函数。

map类对每一行数据进行拆分，当找到我们感兴趣的“kkk”或“nnn”时，就生成一条输出的记录（emit函数的作用）；recude类对map的数据进行汇总，这里只是简单地计数，所以每次+1。

文章来源：

（3）有了代码，我们接着就要编写相应的Makefile了。我的Makefile如下：

HADOOP_INSTALL = /usr/local/hadoop

INCLUDE_PATH = $(HADOOP_INSTALL)/src/c++/

 

CC              = g++

CXXFLAGS = -Wall -g \

                  -I${INCLUDE_PATH}pipes/api/hadoop \

                  -I${INCLUDE_PATH}utils/api/hadoop

LDFLAGS = -ljvm -lhadooppipes -lhadooputils -lpthread

 

OBJECTS=dz_count.o

dz_count: $(OBJECTS)

$(CC) $(CXXFLAGS) -o $@ $(OBJECTS) $(LDFLAGS)

 

其中，HADOOP_INSTALL是你的hadoop安装路径，其余的 INCLUDE_PATH 等请对照你的目录做相应更改，最后生成的可执行程序名为dz_count。这里没有考虑release，因为仅作简单的说明用。

文章来源：

（4）有了代码和Makefile，就可以编译了。编译得到可执行程序dz_count。将其上传到hdfs中：

hadoop fs -put dz_count /my_dir/

其中 “/my_dir/” 是你在hdfs中的目录。

文章来源：

（5）下面就可以运行我们的hadoop程序了：

hadoop pipes -D hadoop.pipes.java.recordreader=true -D hadoop.pipes.java.recordwriter=true -input /data/ -output /my_dir/output -program /my_dir/dz_count

其中，-input /data/ 表明你的输入数据（即你的源数据）所处的hdfs目录为 /data/，-output /my_dir/output 表明你的输出文件目录为 /my_dir/output，“output” 这一级目录必须不存在（如果存在会报错），程序运行时会生成它。-program /my_dir/dz_count 表明你要运行的程序为 /my_dir/ 目录下的 dz_count 程序。

回车之后程序就开始执行，随后你可以在命令行下看到它的状态在更新，或者在hadoop JobTracker中也可以观察到程序的运行状态。

文章来源：

（6）等程序执行完后，如果任务没有失败的话，我们可以看到，你前面指定的hdfs输出目录 /my_dir/output 里生成了一个文件（假设其名为“part-00000”），我们就可以查看执行结果了：

hadoop fs -cat /my_dir/output/part-00000

输出结果形为：

kkk   178099387

nnn   678219805

表明第二项为“kkk”的数据行共有178099387条，而“nnn”则为678219805条。

文章来源：

顺便再说一点废话：

（1）如何中止一个hadoop任务？当你在命令行下提交了一个hadoop job后，就算你按Ctrl+C，也不能中止掉那个job，因为它已经被Jobtracker接管了。这时，你要用如下命令中止它：

hadoop job -kill Job_ID

其中，Job_ID就是你提交的job的ID，可以在Jobtracker中查看到。

（2）一些基本概念：

map-reduce过程中，在map时，hadoop会将输入的数据按一定的大小（例如100M，这个值是可以配置的）分为若干块来处理，一个块对应一个map类，也就是说，一个块只会执行map类的构造函数一次。而每一行记录则对应一个map()方法，也就是说，一行记录就会执行一次map()方法。因此，如果你有什么信息需要输出（例如std::cout）的话，就要注意了：如果在map()方法中输出，则当输入数据量很大时，可能就会输出太多的信息，如果可以在map的构造函数中输出的话，则输出的信息会少得多。

在reduce时，对map输出的同一个key，有一个reduce类，也就是说，无论你的同一个key有多少个value，在reduce的时候只要是同一个key，就会出现在同一个reduce类里，在这个类里的reduce方法中，你用 while (context.nextValue()) 循环可以遍历所有的value，这样就可以处理同一个key的N个value了。 正因为在默认情况下，相同key的记录会落到同一个reducer中，所以，当你的key的数量比你设置的reducer的数量要少的时候，就导致了某些reducer分配不到任何数据，最终输出的某些文件（part-r-xxxxx）是空文件。如果你设置的reducer数量要少于key的数量（这是最常见的情况），那么就会有多个key落入同一个reducer中被处理，但是，每一次reduce()方法被调用时，其中将只包含一个key，同一个reducer里的多个key就会导致reduce()方法被多次调用。

文章来源：

这样，我们就完成了一个完整的C++ hadoop分布式应用程序的编写。