zoukankan      html  css  js  c++  java
  • hadoop(c++实现)

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

    文章来源:http://www.codelast.com/

    (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的详细信息中看到找不到动态库的错误提示。

    文章来源:http://www.codelast.com/

    (2)下面来看看程序:

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

    文章来源:http://www.codelast.com/

    (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,因为仅作简单的说明用。

    文章来源:http://www.codelast.com/

    (4)有了代码和Makefile,就可以编译了。编译得到可执行程序dz_count。将其上传到hdfs中:

    hadoop fs -put dz_count /my_dir/

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

    文章来源:http://www.codelast.com/

    (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中也可以观察到程序的运行状态。

    文章来源:http://www.codelast.com/

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

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

    输出结果形为:

    kkk   178099387

    nnn   678219805

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

    文章来源:http://www.codelast.com/

    顺便再说一点废话:

    (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()方法被多次调用。

    文章来源:http://www.codelast.com/

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

  • 相关阅读:
    AUDIT审计的一些使用
    HOW TO PERFORM BLOCK MEDIA RECOVERY (BMR) WHEN BACKUPS ARE NOT TAKEN BY RMAN. (Doc ID 342972.1)
    使用BBED理解和修改Oracle数据块
    Using Class of Secure Transport (COST) to Restrict Instance Registration in Oracle RAC [ID 1340831.1]
    调试利器GDB概念
    第4章 思科IOS
    第3章 ip地址和子网划分
    第2章 TCPIP
    2020年阅读过的黑客资源推荐篇
    第1章 计算机网络
  • 原文地址:https://www.cnblogs.com/ghost240/p/2526681.html
Copyright © 2011-2022 走看看