假设你有上百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分布式应用程序的编写。