Hadoop基本流程与应用开发

作者 岑文初 发布于 2008年8月13日

领域

语言 & 开发,

架构 & 设计,

运维 & 基础架构,

企业架构

主题

集群与缓存 ,

Java ,

网格计算

标签

Hadoop

── 分布式计算开源框架Hadoop入门实践（三）

Hadoop基本流程

一个图片太大了，只好分割成为两部分。根据流程图来说一下具体一个任务执行的情况。

1. 在分布式环境中客户端创建任务并提交。
2. InputFormat做Map前的预处理，主要负责以下工作：
   
   1. 验证输入的格式是否符合JobConfig的输入定义，这个在实现Map和构建Conf的时候就会知道，不定义可以是Writable的任意子类。
   2. 将input的文件切分为逻辑上的输入InputSplit，其实这就是在上面提到的在分布式文件系统中blocksize是有大小限制的，因此大文件会被划分为多个block。
   3. 通过RecordReader来再次处理inputsplit为一组records，输出给Map。（inputsplit只是逻辑切分的第一步，但是如何根据文件中的信息来切分还需要RecordReader来实现，例如最简单的默认方式就是回车换行的切分）
3. RecordReader处理后的结果作为Map的输入，Map执行定义的Map逻辑，输出处理后的key和value对应到临时中间文件。
4. Combiner可选择配置，主要作用是在每一个Map执行完分析以后，在本地优先作Reduce的工作，减少在Reduce过程中的数据传输量。
5. Partitioner可选择配置，主要作用是在多个Reduce的情况下，指定Map的结果由某一个Reduce处理，每一个Reduce都会有单独的输出文件。（后面的代码实例中有介绍使用场景）
6. Reduce执行具体的业务逻辑，并且将处理结果输出给OutputFormat。
7. OutputFormat的职责是，验证输出目录是否已经存在，同时验证输出结果类型是否如Config中配置，最后输出Reduce汇总后的结果。

业务场景和代码范例

业务场景描述：可设定输入和输出路径（操作系统的路径非HDFS路径），根据访问日志分析某一个应用访问某一个API的总次数和总流量，统计后分别输出到两个文件中。这里仅仅为了测试，没有去细分很多类，将所有的类都归并于一个类便于说明问题。

测试代码类图

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

测试的数据结果很稳定，基本测几次同样条件下都是一样。通过测试结果可以看出以下几点：

1. 机器数对于性能还是有帮助的（等于没说^_^）。
2. 文件复制数的增加只对安全性有帮助，但是对于性能没有太多帮助。而且现在采取的是将操作系统文件拷贝到HDFS中，所以备份多了，准备的时间很长。
3. blocksize对于性能影响很大，首先如果将block划分的太小，那么将会增加job的数量，同时也增加了协作的代价，降低了性能，但是配置的太大也会让job不能最大化并行处理。所以这个值的配置需要根据数据处理的量来考虑。
4. 最后就是除了这个表里面列出来的结果，应该去仔细看输出目录中的_logs/history中的xxx_analysisjob这个文件，里面记录了全部的执行过程以及读写情况。这个可以更加清楚地了解哪里可能会更加耗时。

随想

“云计算”热的烫手，就和SAAS、Web2及SNS等一样，往往都是在搞概念，只有真正踏踏实实的大型互联网公司，才会投入人力物力去研究符合自己的分布式计算。其实当你的数据量没有那么大的时候，这种分布式计算也就仅仅只是一个玩具而已，只有在真正解决问题的过程中，它深层次的问题才会被挖掘出来。

这三篇文章（分布式计算开源框架Hadoop介绍，Hadoop中的集群配置和使用技巧）仅仅是为了给对分布式计算有兴趣的朋友抛个砖，要想真的掘到金子，那么就踏踏实实的去用、去想、去分析。或者自己也会更进一步地去研究框架中的实现机制，在解决自己问题的同时，也能够贡献一些什么。

前几日看到有人跪求成为架构师的方式，看了有些可悲，有些可笑，其实有多少架构师知道什么叫做架构？架构师的职责是什么？与其追求这么一个名号，还不如踏踏实实地做块石头沉到水底。要知道，积累和沉淀的过程就是一种成长。

相关阅读：

1. 分布式计算开源框架Hadoop介绍――分布式计算开源框架Hadoop入门实践（一）。
2. Hadoop中的集群配置和使用技巧――分布式计算开源框架Hadoop入门实践（二）。

---

作者介绍：岑文初，就职于阿里软件公司研发中心平台一部，任架构师。当前主要工作涉及阿里软件开发平台服务框架（ASF）设计与实现，服务集成平台（SIP）设计与实现。没有什么擅长或者精通，工作到现在唯一提升的就是学习能力和速度。个人Blog为：http://blog.csdn.net/cenwenchu79。

志愿参与InfoQ中文站内容建设，请邮件至editors@cn.infoq.com。也欢迎大家到InfoQ中文站用户讨论组参与我们的线上讨论。

1. 返回顶部

   图片是否可以做个链接？

   发表人 zane dennis

   大图片不好放，但是切割后看起来太模糊了，是否可以做个外部链接看完整清晰图？一个小建议

   回复

2. 返回顶部

   受益了

   发表人 lee henry

   学习了!想起以前配HA系统,用Heartbeat的时候,捣鼓了很久,要能有作者这样的入门文章该有多少啊,呵呵!
   
   很欣赏这句话:"与其追求这么一个名号，还不如踏踏实实地做块石头沉到水底。要知道，积累和沉淀的过程就是一种成长。"

   回复

3. 返回顶部

   随想不错

   发表人 feng jarod

   很多谈架构的人 根本见不到大数据 纸上谈兵的居多

   回复

4. 返回顶部

   如果要是引用了第三方的jar，要如何存放才能被引用到？

   发表人 wang haibin

   如题，谢谢！

   回复

5. 返回顶部

   Re: 如果要是引用了第三方的jar，要如何存放才能被引用到？

   发表人 马 士华

   1。把第三方的jar放到每一个Slave的lib里，重起集群，最苯的方法。
   2。把第三方的jar和你的应用打到一个jar文件中。
   3。在提交Job之前使用DistributedCache.setCacheArchives(JarURIs, jobConf);
   把你的第三方的jar Cache到HDFS中。TaskRunner自动会把第三方的jar添加到应用环境中。
   
   www.hadoop.org.cn

   回复

6. 返回顶部

   Re: 如果要是引用了第三方的jar，要如何存放才能被引用到？

   发表人 马 士华

   忘了说一声在使用DistributedCache.setCacheArchives(JarURIs, jobConf);
   后要添加DistributedCache.addCacheArchive(JarURIs, job);
   请看DistributedCache文档和TaskRunner源码就明白了。
   
   www.hadoop.org.cn

   回复

7. 返回顶部

   有点难的说

   发表人 zheng spell

   目前的设计用不到啊用不到

   回复

• 更早的 >