Hadoop学习--Mapper Reduce--day08

haoman

　　mapper类的代码：
　　实现Mapper类的方法
　　import java.io.IOException;
　　import org.apache.hadoop.io.IntWritable;
　　import org.apache.hadoop.io.LongWritable;
　　import org.apache.hadoop.io.Text;
　　import org.apache.hadoop.mapreduce.Mapper;

　　public>　　// constant
　　private static final int MISSING = 9999;
　　// map function
　　@Override
　　protected void map(LongWritable key, Text value, Mapper.Context context)
　　throws IOException, InterruptedException {
　　// per line
　　String line = value.toString();
　　// get year
　　String year = line.substring(15, 19);
　　// get airtemp
　　int airTemperature;
　　if (line.charAt(87) == '+') {
　　airTemperature = Integer.parseInt(line.substring(88, 92));
　　} else {
　　airTemperature = Integer.parseInt(line.substring(87, 92));
　　}
　　// valid air temp data
　　String quality = line.substring(92, 93);
　　if (airTemperature != MISSING && quality.matches("[01459]")) {
　　context.write(new Text(year), new IntWritable(airTemperature));
　　}
　　}
　　}
　　Reduce类的代码：
　　实现Reducer类的方法
　　import java.io.IOException;
　　import org.apache.hadoop.io.IntWritable;
　　import org.apache.hadoop.io.Text;
　　import org.apache.hadoop.mapreduce.Reducer;

　　public>　　@Override
　　protected void reduce(Text key, Iterable values, Context context)
　　throws IOException, InterruptedException {
　　// max
　　int maxValue = Integer.MIN_VALUE;
　　// for
　　for (IntWritable value : values) {
　　maxValue = Math.max(maxValue, value.get());
　　}
　　// output
　　context.write(key, new IntWritable(maxValue));
　　}
　　}
　　主方法的代码：
　　import mapper类实现.MyMapper;
　　import reducer类实现.MyReducer;
　　import org.apache.hadoop.fs.Path;
　　import org.apache.hadoop.io.IntWritable;
　　import org.apache.hadoop.io.Text;
　　import org.apache.hadoop.mapreduce.Job;
　　import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
　　import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

　　public>　　/**
　　* @param args
　　* @throws Exception
　　* @throws IllegalArgumentException
　　*/
　　public static void main(String[] args) throws IllegalArgumentException, Exception {
　　// new job
　　Job job = Job.getInstance();

　　// find jar by>　　job.setJarByClass(MyMapper.class);
　　// job name
　　job.setJobName("Max temperature");
　　FileInputFormat.addInputPath(job, new Path("file:///mnt/hgfs/test-ncdc-data"));
　　FileOutputFormat.setOutputPath(job, new Path("file:///home/hadoop/mr/"));
　　job.setMapperClass(MyMapper.class);
　　job.setReducerClass(MyReducer.class);
　　job.setOutputKeyClass(Text.class);
　　job.setOutputValueClass(IntWritable.class);
　　System.exit(job.waitForCompletion(true) ? 0 : 1);
　　}
　　}