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

Hadoop基本流程

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

• 在分布式环境中客户端创建任务并提交。
• 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的总次数和总流量，统计后分别输出到两个文件中。这里仅仅为了测试，没有去细分很多类，将所有的类都归并于一个类便于说明问题。

测试代码类图

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数处理记录数(万条)执行时间（秒）
29538
2950337
49524
4950178
69521
6950114

Blocksize 5M

Slave数处理记录数(万条)执行时间（秒）
2（文件复制数为1）950337
2（文件复制数为3）950339
6（文件复制数为1）950114
6（文件复制数为3）950117

文件复制数为1

Slave数处理记录数(万条)执行时间（秒）
6(blocksize 5M)9521
6(blocksize 77M)9526
4(blocksize 5M)950178
4(blocksize 50M)95054
6(blocksize 5M)950114
6(blocksize 50M)95044
6(blocksize 77M)95074

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

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