Hadoop学习--Mapper Reduce--day08

opiuytr1

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 class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        // constant
        private static final int MISSING = 9999;

        // map function
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.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 class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

        @Override
        protected void reduce(Text key, Iterable<IntWritable> 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 class MyMapperApp {

        /**
         * @param args
         * @throws Exception
         * @throws IllegalArgumentException
         */
        public static void main(String[] args) throws IllegalArgumentException, Exception {
                // new job
                Job job = Job.getInstance();
                // find jar by ClassName
                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);
        }

}