Hadoop 实战实例
| 阿里云国内75折 回扣 微信号：monov8 |
| 阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6 |
Hadoop 是Google MapReduce的一个Java实现。MapReduce是一种简化的分布式编程模式让程序自动分布到一个由普通机器组成的超大集群上并发执行。就如同java程序员可以不考虑内存泄露一样 MapReduce的run-time系统会解决输入数据的分布细节跨越机器集群的程序执行调度处理机器的失效并且管理机器之间的通讯请求。这样的模式允许程序员可以不需要有什么并发处理或者分布式系统的经验就可以处理超大的分布式系统得资源。
一、概论
作为Hadoop程序员他要做的事情就是
1、定义Mapper处理输入的Key-Value对输出中间结果。
2、定义Reducer可选对中间结果进行规约输出最终结果。
3、定义InputFormat 和OutputFormat可选InputFormat将每行输入文件的内容转换为Java类供Mapper函数使用不定义时默认为String。
4、定义main函数在里面定义一个Job并运行它。
然后的事情就交给系统了。
1.基本概念Hadoop的HDFS实现了google的GFS文件系统NameNode作为文件系统的负责调度运行在 masterDataNode运行在每个机器上。同时Hadoop实现了Google的MapReduceJobTracker作为 MapReduce的总调度运行在masterTaskTracker则运行在每个机器上执行Task。
2.main()函数创建JobConf定义MapperReducerInput/OutputFormat 和输入输出文件目录最后把Job提交給JobTracker等待Job结束。
3.JobTracker创建一个InputFormat的实例调用它的getSplits()方法把输入目录的文件拆分成FileSplist作为Mapper task 的输入生成Mapper task加入Queue。
4.TaskTracker 向 JobTracker索求下一个Map/Reduce。
Mapper Task先从InputFormat创建RecordReader循环读入FileSplits的内容生成Key与Value传给Mapper函数处理完后中间结果写成SequenceFile.
Reducer Task 从运行Mapper的TaskTracker的Jetty上使用http协议获取所需的中间内容33%Sort/Merge后66%执行Reducer函数最后按照OutputFormat写入结果目录。
TaskTracker 每10秒向JobTracker报告一次运行情况每完成一个Task10秒后就会向JobTracker索求下一个Task。
Nutch项目的全部数据处理都构建在Hadoop之上详见Scalable Computing with Hadoop。
二、程序员编写的代码
可以查看hadoop-examples-0.20.203.0.jar,里面也有一个类grep
我们做一个简单的分布式的Grep简单对输入文件进行逐行的正则匹配如果符合就将该行打印到输出文件。因为是简单的全部输出所以我们只要写Mapper函数不用写Reducer函数也不用定义Input/Output Format。
package demo.hadoop
public class HadoopGrep {
public static class RegMapper extends MapReduceBase implements Mapper {
private Pattern pattern;
public void configure(JobConf job) {
pattern = Pattern.compile(job.get( " mapred.mapper.regex "));
}
public void map(WritableComparable key, Writable value, OutputCollector output, Reporter reporter)
throws IOException {
String text = ((Text) value).toString();
Matcher matcher = pattern.matcher(text);
if (matcher.find()) {
output.collect(key, value);
}
}
}
private HadoopGrep () {
} // singleton
public static void main(String[] args) throws Exception {
JobConf grepJob = new JobConf(HadoopGrep. class );
grepJob.setJobName( " grep-search " );
grepJob.set( " mapred.mapper.regex " , args[ 2 ]);
grepJob.setInputPath( new Path(args[ 0 ]));
grepJob.setOutputPath( new Path(args[ 1 ]));
grepJob.setMapperClass(RegMapper. class );
grepJob.setReducerClass(IdentityReducer. class );
JobClient.runJob(grepJob);
}
} RegMapper类的configure()函数接受由main函数传入的查找字符串map() 函数进行正则匹配key是行数value是文件行的内容符合的文件行放入中间结果。
main()函数定义由命令行参数传入的输入输出目录和匹配字符串Mapper函数为RegMapper类Reduce函数是什么都不做直接把中间结果输出到最终结果的的IdentityReducer类运行Job。
整个代码非常简单丝毫没有分布式编程的任何细节。
三.运行Hadoop程序
Hadoop这方面的文档写得不全面综合参考GettingStartedWithHadoop 与Nutch Hadoop Tutorial 两篇后再碰了很多钉子才终于完整的跑起来了记录如下
3.1 local运行模式
完全不进行任何分布式计算不动用任何namenode,datanode的做法适合一开始做调试代码。
解压hadoop其中conf目录是配置目录hadoop的配置文件在hadoop-default.xml如果要修改配置不是直接修改该文件而是修改hadoop-site.xml将该属性在hadoop-site.xml里重新赋值。
hadoop-default.xml的默认配置已经是local运行不用任何修改配置目录里唯一必须修改的是hadoop-env.sh 里JAVA_HOME的位置。
将编译好的HadoopGrep与RegMapper.class 放入hadoop/build/classes/demo/hadoop/目录
或者编译成jar包HadoopGrep.jar放入hadoop/build/classes/demo/hadoop/目录
找一个比较大的xx.log文件放然后运行
bin/hadoop demo.hadoop.HadoopGrep input /tmp/out "[a-b]"
(jar包运行bin/hadoop jar HadoopGrep.jar HadoopGrep input /tmp/output "[a-b]" )
说明
input 为xx.log文件所在目录
/tmp/output为输出目录
"[a-b]" grep的字符串
查看输出目录的结果查看hadoop/logs/里的运行日志。
在重新运行前先删掉输出目录。
3.2 集群运行模式
查看集群配置http://blog.csdn.net/hguisu/article/details/7237395
1 执行bin/hadoop dfs 可以看到它所支持的文件操作指令。
2 创建目录输入inpu
$ bin/hadoop dfs -mkdir input
3上传文件xx.log到指定目录 input
$ bin/hadoop dfs -put xx.log input
4 ) 执行 bin/hadoop demo.hadoop.HadoopGrep input output
(jar包运行bin/hadoop jar HadoopGrep.jar HadoopGrep input /tmp/output "[a-b]" )
5 查看输出文件
将输出文件从分布式文件系统拷贝到本地文件系统查看
$ bin/hadoop fs -get output output
$ cat output/*
或者
在分布式文件系统上查看输出文件
$ bin/hadoop fs -cat output/*
重新执行前,运行hadoop/bin/hadoop dfs -rm output删除output目录
7.运行hadoop/bin/stop-all.sh 结束。
四、效率
经测试Hadoop并不是万用灵丹很取决于文件的大小和数量处理的复杂度以及群集机器的数量相连的带宽当以上四者并不大时hadoop优势并不明显。
比如不用hadoop用java写的简单grep函数处理100M的log文件只要4秒用了hadoop local的方式运行是14秒用了hadoop单机集群的方式是30秒用双机集群10M网口的话更慢慢到不好意思说出来的地步。