Hadoop MapReduce 概念和基础

Hadoop MapReduce 逻辑代码

public class NewMaxTemperature {

    public static final int CONST_INT_ZERO = 0;
    public static final int CONST_INT_ONE = 1;

     //【关键代码】Mapper类继承 Mapper类
    static class NewMaxTemperatureMapper extends Mapper{
       //【关键代码】覆盖父类Mapper的map(LongWritable, Text, Context)方法
        @Override
       protected void map(LongWritable key, Text value, Context context ) throws IOException , InterruptedException{
           //[业务代码]
            super.map(key, value, context); 

            String line = value.toString();
            String year = line.substring(15, 19);
            int airTemperature;
            if( line.charAt(87) == '+'){
                airTemperature = Integer.parseInt(line.substring(88, 92));
            }else{
                airTemperature =Integer.parseInt( line.substring( 87, 92));
            }
            String quality = line.substring(92, 93);
            if(airTemperature != MISSING && quality.matches( "[01459]" )){
                context.write( new Text( year), new IntWritable( airTemperature));
            }
            //[/业务代码]       
        }
   }
   //【关键代码】继承Reducer
   static class NewMaxTemperatureReducer extends Reducer{
       //【关键代码】覆盖Reducer的reduce(Text , Iterable, Context)方法
        @Override
       protected void reduce (Text key, Iterable values, Context context) throws IOException , InterruptedException{
            //[业务代码]
            for (IntWritable value : values){
                maxValue = Math.max(maxValue , value.get() );
            }
            context.write(key,  new IntWritable(maxValue));
            //[/业务代码]
        }

   }
   public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException{
       //【关键代码】创建一个任务，设置任务入口类
       Job job= new Job();
       job.setJarByClass(NewMaxTemperature.class);

       FileInputFormat.addInputPath(job, new Path(args[0]));
       FileInputFormat.addInputPath(job, new Path(args[0]));
        //【关键代码】设置任务的map任务，和reduce任务，及入口类型
       job.setMapperClass(NewMaxTemperatureMapper.class);
       job.setReducerClass(NewMaxTemperatureReducer.class);

       job.setOutputKeyClass(Text.class);
       job.setOutputValueClass(IntWritable.class);
        //【关键代码】 等待完成
       System.exit(job.waitForCompletion(true) ? CONST_INT_ZERO : CONST_INT_ONE);
   }
}

MapReduce作业时客户端执行的单位：包括输入数据、MapReduce、程序和配置信息。 Hadoop把作业分为若干个小任务task， 任务包括map任务和reduce任务。
有两种节点控制作业的完成， 这种节点包括jobtracker和多个tasktracker。jobtracker是tasktracker的调度器， tasktracker是jobtracker的slaves。tasktracker在运行过程中从jobtracker获取任务，并把进度报告给jobtracker， 如果一个任务失败，jobtracker会重新调度任务到另外一个tasktracker。 Hadoop 把输入的数据划分为小数据到MapReduce，称为输入分片（input split），每个分片包括创建map任务， 由它运行自定义的map函数来分析片中的记录。分片大小一般为一个block的大小64M。

[img]http://dl.iteye.com/upload/attachment/370088/e37e019c-c9a8-39fc-ae3a-8b7b5599f5c7.jpg" alt="[/img]

map任务把输出到本地硬盘， 而不是HDFS，应为map的输出是一个中间输出，中间输出是reduce处理的输入， 最终结果是reduce输出的。一但结果完成，map的输出是可以删除的，所以map的结果一般是不会存储在HDFS中的。

[img]http://dl.iteye.com/upload/attachment/370101/2f8579f7-bbab-36a2-8b61-f9692246315e.jpg" alt="[/img]

reduce任务不具备本地读取的优势-- 一个单一的reduce任务的输入往往来自于所有map的输出。 map的输出通过网络传输到reduce任务运行的节点， 并在哪里合并， 然后传递给自定义的reduce函数， 为了增加可靠性， reduce的输出通常存储在HDFS中。
[img]http://dl.iteye.com/upload/attachment/370096/75086578-e367-3d3b-bf0f-98b85de34661.jpg" alt="[/img]
如果是多个reducer， map任务会会对输出结果进行分区， 为配给reduce任务创建一个分区（partition）。 每个分区包含许多键/值对， 分区可以有用户自定义的partitioner来控制，通常是采用默认的分区工具， 它使用的算法是hashhas函数， 这种方法效率非常高。
也有可能不存在reduce任务， 直接的输出就是map任务的输出
[img]http://dl.iteye.com/upload/attachment/370094/85e96ac9-a17a-37d6-b046-13e1af4c6850.jpg" alt="[/img]


注释： 大部分内容来自《Hadoop权威指南》