Hadoop学习之MapReduce执行过程详解

  1 package com.abc.yarn;
  2  
  3 import java.io.IOException;
  4  
  5 import org.apache.hadoop.conf.Configuration;
  6 import org.apache.hadoop.fs.Path;
  7 import org.apache.hadoop.io.IntWritable;
  8 import org.apache.hadoop.io.LongWritable;
  9 import org.apache.hadoop.io.Text;
 10 import org.apache.hadoop.mapreduce.Job;
 11 import org.apache.hadoop.mapreduce.Mapper;
 12 import org.apache.hadoop.mapreduce.Reducer;
 13 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 14 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 15  
 16 public class Temperature {
 17     /**
 18      * 四个泛型类型分别代表：
 19      * KeyIn        Mapper的输入数据的Key，这里是每行文字的起始位置（0,11,...）
 20      * ValueIn      Mapper的输入数据的Value，这里是每行文字
 21      * KeyOut       Mapper的输出数据的Key，这里是每行文字中的“年份”
 22      * ValueOut     Mapper的输出数据的Value，这里是每行文字中的“气温”
 23      */
 24     static class TempMapper extends
 25             Mapper {
 26         @Override
 27         public void map(LongWritable key, Text value, Context context)
 28                 throws IOException, InterruptedException {
 29             // 打印样本: Before Mapper: 0, 2000010115
 30             System.out.print("Before Mapper: " + key + ", " + value);
 31             String line = value.toString();
 32             String year = line.substring(0, 4);
 33             int temperature = Integer.parseInt(line.substring(8));
 34             context.write(new Text(year), new IntWritable(temperature));
 35             // 打印样本: After Mapper:2000, 15
 36             System.out.println(
 37                     "======" +
 38                     "After Mapper:" + new Text(year) + ", " + new IntWritable(temperature));
 39         }
 40     }
 41  
 42     /**
 43      * 四个泛型类型分别代表：
 44      * KeyIn        Reducer的输入数据的Key，这里是每行文字中的“年份”
 45      * ValueIn      Reducer的输入数据的Value，这里是每行文字中的“气温”
 46      * KeyOut       Reducer的输出数据的Key，这里是不重复的“年份”
 47      * ValueOut     Reducer的输出数据的Value，这里是这一年中的“最高气温”
 48      */
 49     static class TempReducer extends
 50             Reducer {
 51         @Override
 52         public void reduce(Text key, Iterable values,
 53                 Context context) throws IOException, InterruptedException {
 54             int maxValue = Integer.MIN_VALUE;
 55             StringBuffer sb = new StringBuffer();
 56             //取values的最大值
 57             for (IntWritable value : values) {
 58                 maxValue = Math.max(maxValue, value.get());
 59                 sb.append(value).append(", ");
 60             }
 61             // 打印样本： Before Reduce: 2000, 15, 23, 99, 12, 22, 
 62             System.out.print("Before Reduce: " + key + ", " + sb.toString());
 63             context.write(key, new IntWritable(maxValue));
 64             // 打印样本： After Reduce: 2000, 99
 65             System.out.println(
 66                     "======" +
 67                     "After Reduce: " + key + ", " + maxValue);
 68         }
 69     }
 70  
 71     public static void main(String[] args) throws Exception {
 72         //输入路径
 73         String dst = "hdfs://localhost:9000/intput.txt";
 74         //输出路径，必须是不存在的，空文件加也不行。
 75         String dstOut = "hdfs://localhost:9000/output";
 76         Configuration hadoopCOnfig= new Configuration();
 77          
 78         hadoopConfig.set("fs.hdfs.impl", 
 79             org.apache.hadoop.hdfs.DistributedFileSystem.class.getName()
 80         );
 81         hadoopConfig.set("fs.file.impl",
 82             org.apache.hadoop.fs.LocalFileSystem.class.getName()
 83         );
 84         Job job = new Job(hadoopConfig);
 85          
 86         //如果需要打成jar运行，需要下面这句
 87         //job.setJarByClass(NewMaxTemperature.class);
 88  
 89         //job执行作业时输入和输出文件的路径
 90         FileInputFormat.addInputPath(job, new Path(dst));
 91         FileOutputFormat.setOutputPath(job, new Path(dstOut));
 92  
 93         //指定自定义的Mapper和Reducer作为两个阶段的任务处理类
 94         job.setMapperClass(TempMapper.class);
 95         job.setReducerClass(TempReducer.class);
 96          
 97         //设置最后输出结果的Key和Value的类型
 98         job.setOutputKeyClass(Text.class);
 99         job.setOutputValueClass(IntWritable.class);
100          
101         //执行job，直到完成
102         job.waitForCompletion(true);
103         System.out.println("Finished");
104     }
105 }

 1 Before Mapper: 0, 2014010114======After Mapper:2014, 14
 2 Before Mapper: 11, 2014010216======After Mapper:2014, 16
 3 Before Mapper: 22, 2014010317======After Mapper:2014, 17
 4 Before Mapper: 33, 2014010410======After Mapper:2014, 10
 5 Before Mapper: 44, 2014010506======After Mapper:2014, 6
 6 Before Mapper: 55, 2012010609======After Mapper:2012, 9
 7 Before Mapper: 66, 2012010732======After Mapper:2012, 32
 8 Before Mapper: 77, 2012010812======After Mapper:2012, 12
 9 Before Mapper: 88, 2012010919======After Mapper:2012, 19
10 Before Mapper: 99, 2012011023======After Mapper:2012, 23
11 Before Mapper: 110, 2001010116======After Mapper:2001, 16
12 Before Mapper: 121, 2001010212======After Mapper:2001, 12
13 Before Mapper: 132, 2001010310======After Mapper:2001, 10
14 Before Mapper: 143, 2001010411======After Mapper:2001, 11
15 Before Mapper: 154, 2001010529======After Mapper:2001, 29
16 Before Mapper: 165, 2013010619======After Mapper:2013, 19
17 Before Mapper: 176, 2013010722======After Mapper:2013, 22
18 Before Mapper: 187, 2013010812======After Mapper:2013, 12
19 Before Mapper: 198, 2013010929======After Mapper:2013, 29
20 Before Mapper: 209, 2013011023======After Mapper:2013, 23
21 Before Mapper: 220, 2008010105======After Mapper:2008, 5
22 Before Mapper: 231, 2008010216======After Mapper:2008, 16
23 Before Mapper: 242, 2008010337======After Mapper:2008, 37
24 Before Mapper: 253, 2008010414======After Mapper:2008, 14
25 Before Mapper: 264, 2008010516======After Mapper:2008, 16
26 Before Mapper: 275, 2007010619======After Mapper:2007, 19
27 Before Mapper: 286, 2007010712======After Mapper:2007, 12
28 Before Mapper: 297, 2007010812======After Mapper:2007, 12
29 Before Mapper: 308, 2007010999======After Mapper:2007, 99
30 Before Mapper: 319, 2007011023======After Mapper:2007, 23
31 Before Mapper: 330, 2010010114======After Mapper:2010, 14
32 Before Mapper: 341, 2010010216======After Mapper:2010, 16
33 Before Mapper: 352, 2010010317======After Mapper:2010, 17
34 Before Mapper: 363, 2010010410======After Mapper:2010, 10
35 Before Mapper: 374, 2010010506======After Mapper:2010, 6
36 Before Mapper: 385, 2015010649======After Mapper:2015, 49
37 Before Mapper: 396, 2015010722======After Mapper:2015, 22
38 Before Mapper: 407, 2015010812======After Mapper:2015, 12
39 Before Mapper: 418, 2015010999======After Mapper:2015, 99
40 Before Mapper: 429, 2015011023======After Mapper:2015, 23
41 Finished