TemperatureMR.java
1package cn.kissoft.hadoop.week05; 2 3import org.apache.hadoop.fs.Path; 4import org.apache.hadoop.io.IntWritable; 5import org.apache.hadoop.io.Text; 6import org.apache.hadoop.mapreduce.Job; 7import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 8import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 9 10public class TemperatureMR { 11 12 public static void main(String[] args) throws Exception { 13 if (args.length != 2) { 14 System.err.println("Usage: Temperature <input path> <output path>"); 15 System.exit(-1); 16 } 17 Job job = new Job(); 18 job.setJarByClass(TemperatureMR.class); 19 job.setJobName("Max and min temperature"); 20 FileInputFormat.addInputPath(job, new Path(args[0])); 21 FileOutputFormat.setOutputPath(job, new Path(args[1])); 22 job.setMapperClass(TemperatureMapper.class); 23 job.setReducerClass(TemperatureReducer.class); 24 job.setOutputKeyClass(Text.class); 25 job.setOutputValueClass(IntWritable.class); 26 System.exit(job.waitForCompletion(true) ? 0 : 1); 27 } 28}
TemperatureMapper.java
1package cn.kissoft.hadoop.week05; 2 3import java.io.IOException; 4 5import org.apache.hadoop.io.IntWritable; 6import org.apache.hadoop.io.LongWritable; 7import org.apache.hadoop.io.Text; 8import org.apache.hadoop.mapreduce.Mapper; 9 10public class TemperatureMapper extends Mapper<LongWritable, Text, Text, IntWritable> { 11 private static final int MISSING = 9999; 12 13 @Override 14 public void map(LongWritable key, Text value, Context context) 15 throws IOException, InterruptedException { 16 String line = value.toString(); 17 String year = line.substring(0, 4); 18 int airTemperature = Integer.parseInt(line.substring(13, 19).trim()); 19 if (Math.abs(airTemperature) != MISSING) { 20 context.write(new Text(year), new IntWritable(airTemperature)); 21 } 22 } 23}
MaxTemperatureReducer.java
1package cn.kissoft.hadoop.week05; 2 3import java.io.IOException; 4 5import org.apache.hadoop.io.IntWritable; 6import org.apache.hadoop.io.Text; 7import org.apache.hadoop.mapreduce.Reducer; 8 9public class TemperatureReducer extends Reducer<Text, IntWritable, Text, IntWritable> { 10 11 @Override 12 public void reduce(Text key, Iterable<IntWritable> values, Context context) 13 throws IOException, InterruptedException { 14 15 int maxValue = Integer.MIN_VALUE; 16 int minValue = Integer.MAX_VALUE; 17 for (IntWritable value : values) { 18 maxValue = Math.max(maxValue, value.get()); 19 minValue = Math.min(minValue, value.get()); 20 } 21 context.write(key, new IntWritable(maxValue)); 22 context.write(key, new IntWritable(minValue)); 23 } 24}
运行过程
1[wukong@bd11 guide]$ hadoop jar pc.jar cn.kissoft.hadoop.week05.TemperatureMR ./ch02/1959.txt ./ch02/out/ 2Warning: $HADOOP_HOME is deprecated. 3 4 514/08/15 16:29:32 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. 614/08/15 16:29:32 INFO input.FileInputFormat: Total input paths to process : 1 714/08/15 16:29:32 INFO util.NativeCodeLoader: Loaded the native-hadoop library 814/08/15 16:29:32 WARN snappy.LoadSnappy: Snappy native library not loaded 914/08/15 16:29:34 INFO mapred.JobClient: Running job: job_201408151617_0003 1014/08/15 16:29:35 INFO mapred.JobClient: map 0% reduce 0% 1114/08/15 16:29:47 INFO mapred.JobClient: map 100% reduce 0% 1214/08/15 16:30:00 INFO mapred.JobClient: map 100% reduce 100% 1314/08/15 16:30:04 INFO mapred.JobClient: Job complete: job_201408151617_0003 1414/08/15 16:30:04 INFO mapred.JobClient: Counters: 29 1514/08/15 16:30:04 INFO mapred.JobClient: Job Counters 1614/08/15 16:30:04 INFO mapred.JobClient: Launched reduce tasks=1 1714/08/15 16:30:04 INFO mapred.JobClient: SLOTS_MILLIS_MAPS=14989 1814/08/15 16:30:04 INFO mapred.JobClient: Total time spent by all reduces waiting after reserving slots (ms)=0 1914/08/15 16:30:04 INFO mapred.JobClient: Total time spent by all maps waiting after reserving slots (ms)=0 2014/08/15 16:30:04 INFO mapred.JobClient: Launched map tasks=1 2114/08/15 16:30:04 INFO mapred.JobClient: Data-local map tasks=1 2214/08/15 16:30:04 INFO mapred.JobClient: SLOTS_MILLIS_REDUCES=12825 2314/08/15 16:30:04 INFO mapred.JobClient: File Output Format Counters 2414/08/15 16:30:04 INFO mapred.JobClient: Bytes Written=19 2514/08/15 16:30:04 INFO mapred.JobClient: FileSystemCounters 2614/08/15 16:30:04 INFO mapred.JobClient: FILE_BYTES_READ=9180486 2714/08/15 16:30:04 INFO mapred.JobClient: HDFS_BYTES_READ=27544475 2814/08/15 16:30:04 INFO mapred.JobClient: FILE_BYTES_WRITTEN=13886908 2914/08/15 16:30:04 INFO mapred.JobClient: HDFS_BYTES_WRITTEN=19 3014/08/15 16:30:04 INFO mapred.JobClient: File Input Format Counters 3114/08/15 16:30:04 INFO mapred.JobClient: Bytes Read=27544368 3214/08/15 16:30:04 INFO mapred.JobClient: Map-Reduce Framework 3314/08/15 16:30:04 INFO mapred.JobClient: Map output materialized bytes=4590240 3414/08/15 16:30:04 INFO mapred.JobClient: Map input records=444264 3514/08/15 16:30:04 INFO mapred.JobClient: Reduce shuffle bytes=4590240 3614/08/15 16:30:04 INFO mapred.JobClient: Spilled Records=1251882 3714/08/15 16:30:04 INFO mapred.JobClient: Map output bytes=3755646 3814/08/15 16:30:04 INFO mapred.JobClient: Total committed heap usage (bytes)=218865664 3914/08/15 16:30:04 INFO mapred.JobClient: CPU time spent (ms)=6280 4014/08/15 16:30:04 INFO mapred.JobClient: Combine input records=0 4114/08/15 16:30:04 INFO mapred.JobClient: SPLIT_RAW_BYTES=107 4214/08/15 16:30:04 INFO mapred.JobClient: Reduce input records=417294 4314/08/15 16:30:04 INFO mapred.JobClient: Reduce input groups=1 4414/08/15 16:30:04 INFO mapred.JobClient: Combine output records=0 4514/08/15 16:30:04 INFO mapred.JobClient: Physical memory (bytes) snapshot=322985984 4614/08/15 16:30:04 INFO mapred.JobClient: Reduce output records=2 4714/08/15 16:30:04 INFO mapred.JobClient: Virtual memory (bytes) snapshot=1455579136 4814/08/15 16:30:04 INFO mapred.JobClient: Map output records=417294
运行结果
1[wukong@bd11 guide]$ hadoop fs -cat ./ch02/out/part-r-00000 2 3Warning: $HADOOP_HOME is deprecated. 4 51959 418 61959 -400
截图
**
**
