Hadoop基本流程与应用开发
http://www.infoq.com/cn/articles/hadoop-process-develop
http://www.infoq.com/cn/articles/hadoop-intro;jsessionid=1B566AFBB2B51A0D70380914DFFC8333
http://www.infoq.com/cn/articles/hadoop-config-tip;jsessionid=1B566AFBB2B51A0D70380914DFFC8333
── 分布式计算开源框架Hadoop入门实践(三)
http://www.infoq.com/resource/articles/hadoop-process-develop/zh/resources/image1.gif;jsessionid=1B566AFBB2B51A0D70380914DFFC8333
http://www.infoq.com/resource/articles/hadoop-process-develop/zh/resources/image2.gif;jsessionid=1B566AFBB2B51A0D70380914DFFC8333
一个图片太大了,只好分割成为两部分。根据流程图来说一下具体一个任务执行的情况。
- 在分布式环境中客户端创建任务并提交。
- InputFormat做Map前的预处理,主要负责以下工作:
- 验证输入的格式是否符合JobConfig的输入定义,这个在实现Map和构建Conf的时候就会知道,不定义可以是Writable的任意子类。
- 将input的文件切分为逻辑上的输入InputSplit,其实这就是在上面提到的在分布式文件系统中blocksize是有大小限制的,因此大文件会被划分为多个block。
- 通过RecordReader来再次处理inputsplit为一组records,输出给Map。(inputsplit只是逻辑切分的第一步,但是如何根据文件中的信息来切分还需要RecordReader来实现,例如最简单的默认方式就是回车换行的切分)
- RecordReader处理后的结果作为Map的输入,Map执行定义的Map逻辑,输出处理后的key和value对应到临时中间文件。
- Combiner可选择配置,主要作用是在每一个Map执行完分析以后,在本地优先作Reduce的工作,减少在Reduce过程中的数据传输量。
- Partitioner可选择配置,主要作用是在多个Reduce的情况下,指定Map的结果由某一个Reduce处理,每一个Reduce都会有单独的输出文件。(后面的代码实例中有介绍使用场景)
- Reduce执行具体的业务逻辑,并且将处理结果输出给OutputFormat。
- OutputFormat的职责是,验证输出目录是否已经存在,同时验证输出结果类型是否如Config中配置,最后输出Reduce汇总后的结果。
业务场景和代码范例
业务场景描述:可设定输入和输出路径(操作系统的路径非HDFS路径),根据访问日志分析某一个应用访问某一个API的总次数和总流量,统计后分别输出到两个文件中。这里仅仅为了测试,没有去细分很多类,将所有的类都归并于一个类便于说明问题。
http://www.infoq.com/resource/articles/hadoop-process-develop/zh/resources/image3.gif;jsessionid=1B566AFBB2B51A0D70380914DFFC8333
测试代码类图
LogAnalysiser就是主类,主要负责创建、提交任务,并且输出部分信息。内部的几个子类用途可以参看流程中提到的角色职责。具体地看看几个类和方法的代码片断:
LogAnalysiser::MapClass
public static class MapClass extends MapReduceBase
implements Mapper<LongWritable, Text, Text, LongWritable>
{
public void map(LongWritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter)
throws IOException
{
String line = value.toString();//没有配置RecordReader,所以默认采用line的实现,key就是行号,value就是行内容
if (line == null || line.equals(""))
return;
String[] words = line.split(",");
if (words == null || words.length < 8)
return;
String appid = words[1];
String apiName = words[2];
LongWritable recbytes = new LongWritable(Long.parseLong(words[7]));
Text record = new Text();
record.set(new StringBuffer("flow::").append(appid)
.append("::").append(apiName).toString());
reporter.progress();
output.collect(record, recbytes);//输出流量的统计结果,通过flow::作为前缀来标示。
record.clear();
record.set(new StringBuffer("count::").append(appid).append("::").append(apiName).toString());
output.collect(record, new LongWritable(1));//输出次数的统计结果,通过count::作为前缀来标示
}
}
LogAnalysiser:: PartitionerClass
public static class PartitionerClass implements Partitioner<Text, LongWritable>
{
public int getPartition(Text key, LongWritable value, int numPartitions)
{
if (numPartitions >= 2)//Reduce 个数,判断流量还是次数的统计分配到不同的Reduce
if (key.toString().startsWith("flow::"))
return 0;
else
return 1;
else
return 0;
}
public void configure(JobConf job){}
}
LogAnalysiser:: CombinerClass
参看ReduceClass,通常两者可以使用一个,不过这里有些不同的处理就分成了两个。在ReduceClass中蓝色的行表示在CombinerClass中不存在。
LogAnalysiser:: ReduceClass
public static class ReduceClass extends MapReduceBase
implements Reducer<Text, LongWritable,Text, LongWritable>
{
public void reduce(Text key, Iterator<LongWritable> values,
OutputCollector<Text, LongWritable> output, Reporter reporter)throws IOException
{
Text newkey = new Text();
newkey.set(key.toString().substring(key.toString().indexOf("::")+2));
LongWritable result = new LongWritable();
long tmp = 0;
int counter = 0;
while(values.hasNext())//累加同一个key的统计结果
{
tmp = tmp + values.next().get();
counter = counter +1;//担心处理太久,JobTracker长时间没有收到报告会认为TaskTracker已经失效,因此定时报告一下
if (counter == 1000)
{
counter = 0;
reporter.progress();
}
}
result.set(tmp);
output.collect(newkey, result);//输出最后的汇总结果
}
}
LogAnalysiser
public static void main(String[] args)
{
try
{
run(args);
} catch (Exception e)
{
e.printStackTrace();
}
}
public static void run(String[] args) throws Exception
{
if (args == null || args.length <2)
{
System.out.println("need inputpath and outputpath");
return;
}
String inputpath = args[0];
String outputpath = args[1];
String shortin = args[0];
String shortout = args[1];
if (shortin.indexOf(File.separator) >= 0)
shortin = shortin.substring(shortin.lastIndexOf(File.separator));
if (shortout.indexOf(File.separator) >= 0)
shortout = shortout.substring(shortout.lastIndexOf(File.separator));
SimpleDateFormat formater = new SimpleDateFormat("yyyy.MM.dd");
shortout = new StringBuffer(shortout).append("-")
.append(formater.format(new Date())).toString();
if (!shortin.startsWith("/"))
shortin = "/" + shortin;
if (!shortout.startsWith("/"))
shortout = "/" + shortout;
shortin = "/user/root" + shortin;
shortout = "/user/root" + shortout;
File inputdir = new File(inputpath);
File outputdir = new File(outputpath);
if (!inputdir.exists() || !inputdir.isDirectory())
{
System.out.println("inputpath not exist or isn't dir!");
return;
}
if (!outputdir.exists())
{
new File(outputpath).mkdirs();
}
JobConf conf = new JobConf(new Configuration(),LogAnalysiser.class);//构建Config
FileSystem fileSys = FileSystem.get(conf);
fileSys.copyFromLocalFile(new Path(inputpath), new Path(shortin));//将本地文件系统的文件拷贝到HDFS中
conf.setJobName("analysisjob");
conf.setOutputKeyClass(Text.class);//输出的key类型,在OutputFormat会检查
conf.setOutputValueClass(LongWritable.class); //输出的value类型,在OutputFormat会检查
conf.setMapperClass(MapClass.class);
conf.setCombinerClass(CombinerClass.class);
conf.setReducerClass(ReduceClass.class);
conf.setPartitionerClass(PartitionerClass.class);
conf.set("mapred.reduce.tasks", "2");//强制需要有两个Reduce来分别处理流量和次数的统计
FileInputFormat.setInputPaths(conf, shortin);//hdfs中的输入路径
FileOutputFormat.setOutputPath(conf, new Path(shortout));//hdfs中输出路径
Date startTime = new Date();
System.out.println("Job started: " + startTime);
JobClient.runJob(conf);
Date end_time = new Date();
System.out.println("Job ended: " + end_time);
System.out.println("The job took " + (end_time.getTime() - startTime.getTime()) /1000 + " seconds.");
//删除输入和输出的临时文件
fileSys.copyToLocalFile(new Path(shortout),new Path(outputpath));
fileSys.delete(new Path(shortin),true);
fileSys.delete(new Path(shortout),true);
}
以上的代码就完成了所有的逻辑性代码,然后还需要一个注册驱动类来注册业务Class为一个可标示的命令,让hadoop jar可以执行。
public class ExampleDriver {
public static void main(String argv[]){
ProgramDriver pgd = new ProgramDriver();
try {
pgd.addClass("analysislog", LogAnalysiser.class, "A map/reduce program that analysis log .");
pgd.driver(argv);
}
catch(Throwable e){
e.printStackTrace();
}
}
}
将代码打成jar,并且设置jar的mainClass为ExampleDriver这个类。在分布式环境启动以后执行如下语句:
hadoop jar analysiser.jar analysislog /home/wenchu/test-in /home/wenchu/test-out
在/home/wenchu/test-in中是需要分析的日志文件,执行后就会看见整个执行过程,包括了Map和Reduce的进度。执行完毕会在/home/wenchu/test-out下看到输出的内容。有两个文件:part-00000和part-00001分别记录了统计后的结果。 如果需要看执行的具体情况,可以看在输出目录下的_logs/history/xxxx_analysisjob,里面罗列了所有的Map,Reduce的创建情况以及执行情况。在运行期也可以通过浏览器来查看Map,Reduce的情况:http://MasterIP:50030/jobtracker.jsp
Hadoop集群测试
首先这里使用上面的范例作为测试,也没有做太多的优化配置,这个测试结果只是为了看看集群的效果,以及一些参数配置的影响。
文件复制数为1,blocksize 5M
Slave数 | 处理记录数(万条) | 执行时间(秒) |
2 | 95 | 38 |
2 | 950 | 337 |
4 | 95 | 24 |
4 | 950 | 178 |
6 | 95 | 21 |
6 | 950 | 114 |
Blocksize 5M
Slave数 | 处理记录数(万条) | 执行时间(秒) |
2(文件复制数为1) | 950 | 337 |
2(文件复制数为3) | 950 | 339 |
6(文件复制数为1) | 950 | 114 |
6(文件复制数为3) | 950 | 117 |
文件复制数为1
Slave数 | 处理记录数(万条) | 执行时间(秒) |
6(blocksize 5M) | 95 | 21 |
6(blocksize 77M) | 95 | 26 |
4(blocksize 5M) | 950 | 178 |
4(blocksize 50M) | 950 | 54 |
6(blocksize 5M) | 950 | 114 |
6(blocksize 50M) | 950 | 44 |
6(blocksize 77M) | 950 | 74 |
测试的数据结果很稳定,基本测几次同样条件下都是一样。通过测试结果可以看出以下几点:
- 机器数对于性能还是有帮助的(等于没说^_^)。
- 文件复制数的增加只对安全性有帮助,但是对于性能没有太多帮助。而且现在采取的是将操作系统文件拷贝到HDFS中,所以备份多了,准备的时间很长。
- blocksize对于性能影响很大,首先如果将block划分的太小,那么将会增加job的数量,同时也增加了协作的代价,降低了性能,但是配置的太大也会让job不能最大化并行处理。所以这个值的配置需要根据数据处理的量来考虑。
- 最后就是除了这个表里面列出来的结果,应该去仔细看输出目录中的_logs/history中的xxx_analysisjob这个文件,里面记录了全部的执行过程以及读写情况。这个可以更加清楚地了解哪里可能会更加耗时。
随想
“云计算”热的烫手,就和SAAS、Web2及SNS等一样,往往都是在搞概念,只有真正踏踏实实的大型互联网公司,才会投入人力物力去研究符合自己的分布式计算。其实当你的数据量没有那么大的时候,这种分布式计算也就仅仅只是一个玩具而已,只有在真正解决问题的过程中,它深层次的问题才会被挖掘出来。
这三篇文章(分布式计算开源框架Hadoop介绍,Hadoop中的集群配置和使用技巧)仅仅是为了给对分布式计算有兴趣的朋友抛个砖,要想真的掘到金子,那么就踏踏实实的去用、去想、去分析。或者自己也会更进一步地去研究框架中的实现机制,在解决自己问题的同时,也能够贡献一些什么。
前几日看到有人跪求成为架构师的方式,看了有些可悲,有些可笑,其实有多少架构师知道什么叫做架构?架构师的职责是什么?与其追求这么一个名号,还不如踏踏实实地做块石头沉到水底。要知道,积累和沉淀的过程就是一种成长。
相关阅读:
- 分布式计算开源框架Hadoop介绍――分布式计算开源框架Hadoop入门实践(一)。
- Hadoop中的集群配置和使用技巧――分布式计算开源框架Hadoop入门实践(二)。
作者介绍:岑文初,就职于阿里软件公司研发中心平台一部,任架构师。当前主要工作涉及阿里软件开发平台服务框架(ASF)设计与实现,服务集成平台(SIP)设计与实现。没有什么擅长或者精通,工作到现在唯一提升的就是学习能力和速度。个人Blog为:http://blog.csdn.net/cenwenchu79。
志愿参与InfoQ中文站内容建设,请邮件至editors@cn.infoq.com。也欢迎大家到InfoQ中文站用户讨论组参与我们的线上讨论。
7 条回复
关注此讨论 回复
Hadoop基本流程与应用开发相关推荐
- 【嵌入式Linux(基础篇)】嵌入式Linux底层系统开发流程和应用开发流程
1.嵌入式Linux系统体系架构 一个完整的嵌入式Linux系统体系架构如图,大概可以分为三步: 硬件PCB板设计 底层系统开发 应用开发 2.硬件PCB设计流程 ① 获取所用芯片芯片手册: ② 建立 ...
- 我的Hadoop安装流程
Hadoop安装流程 目录 Hadoop安装流程... 1 虚拟机安装设置... 1 设置网卡... 1 共享文件夹... 2 虚拟机安装过程... 2 基础环境安装... 4 赋予用户root权限. ...
- 移动app开发流程,app开发步骤
个人github:https://github.com/qiilee 欢迎follow 移动app开发流程,app开发步骤 app是手机软件的简称.手机主流的有ios.andriod,windows ...
- 大数据项目实践:基于hadoop+spark+mongodb+mysql+c#开发医院临床知识库系统
从20世纪90年代数字化医院概念提出到至今的20多年时间,数字化医院(Digital Hospital)在国内各大医院飞速的普及推广发展,并取得骄人成绩.不但有数字化医院管理信息系统(HIS).影像存 ...
- 研发流程在敏捷开发中的详解
在传统的软件研发模型中,从提出需求到最后交付,时间周期较长.瀑布模型遵循需求分析.设计.编码.集成.测试.维护六个步骤进行.一旦需求发生变化,不仅浪费前期投入,还不易于调整. 1. 敏捷开发是什么 在 ...
- 学习笔记Hadoop(十三)—— MapReduce开发入门(1)—— MapReduce开发环境搭建、MapReduce单词计数源码分析
一.MapReduce MapReduce是Google提出的一个软件架构,用于大规模数据集(大于1TB)的并行运算.概念"Map(映射)"和"Reduce(归纳)&qu ...
- 审批流_想做流程审批的开发?带你玩转Activity工作流(一)
现在的很多传统企业转型互联网,最多的"转型"工作之一,就是公司的流程审批模块,如何进行线上运行,实现降本增效. -- 23号老板 01引入 现在的很多传统企业转型互联网, ...
- destoon php os,destoon运行流程二次开发必看
destoon系统采用B/S架构,MVC模式开发.融入了模型化.模板.缓存.AJAX.SEO等前沿技术 下面是运行流程,虽然版本有更新但大体上是没有变化太多的//代码首先包含common.inc.ph ...
- 使用O2OA二次开发搭建企业办公平台(十三)流程开发篇:报销审批流程表单开发...
本博客为O2OA系列教程.O2OA使用手册,教程目录和各章节天梯将在连载完后更新. 使用O2OA二次开发搭建企业办公平台(一)平台部署篇:平台下载和部署 使用O2OA二次开发搭建企业办公平台(二)平台 ...
最新文章
- snort inline模式和passive模式区别
- 鼠标右击 左击 等总结
- leetcood学习笔记-2-两数相加
- Hibernate所用15个jar包
- 《计算机系统:系统架构与操作系统的高度集成》——1.6 操作系统的演化
- 云计算遇上区块链,会产生怎样的能量和火花?
- Pytorch常用技巧记录
- 2021-06-09反射的类型方法
- bootstrap 4 自动水平居中
- 最受欢迎的 50 个数据可视化图表
- 女性黑客的崛起将造福世界
- 行走在网格之间:微博用户关系模型
- 行频、场频与分辨率、刷新率
- html5 audio音乐播放,循环播放等9款播放器
- java mysql点赞功能_用Java做一个类似于微博QQ空间点赞的功能-Fun言
- iOS 客户端基于 WebP 图片格式的流量优化(下)
- oracle v backup,睿备份oracle下载
- thinkpad x1c 2015使用GPT分区方式安装原版win7系统
- 利用pandas进行简单数据分析-医院销售数据分析案例
- 构建三维动画的五大设计软件,你最pick谁?
热门文章
- 《每日一题》738. Monotone Increasing Digits 单调递增的数字
- 2019第十届蓝桥杯C/C++ A组省赛 —— 第一题:平方和
- 大数据WEB阶段 Maven与SSM框架整合
- 大数据WEB阶段(十八)数据库事务
- linux多级菜单脚本教程,linux shell 编写菜单脚本事例
- C++代码片段(三)过滤可变模板参数中某些类型的变量
- 最小栈—leetcode155
- php两数相乘,php – 如何将数字相乘,直到达到一位数字并计算数字?
- 开始体验Kali Linux
- 机器学习入门:机器学习概论