Maximum Temperature Mapreduce Program

DriverClass.Java:-
----------------------

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 DriverClass {

                public static void main(String[] args) throws Exception {

                                if (args.length != 2) {

                                                System.err.println("Usage: MaxTemperature <input path> <output path>");

                                                System.exit(-1);

                                }

                                Job job = new Job();

                                job.setJarByClass(DriverClass.class);

                                job.setJobName("Max temperature");

                                FileInputFormat.addInputPath(job, new Path(args[0]));

                                FileOutputFormat.setOutputPath(job, new Path(args[1]));

                                job.setMapperClass(DMapper.class);

                                job.setCombinerClass(DReducer.class);

                                job.setReducerClass(DReducer.class);

                                job.setOutputKeyClass(Text.class);

                                job.setOutputValueClass(IntWritable.class);

                                System.exit(job.waitForCompletion(true) ? 0 : 1);

                }

 } 

DMapper.Java:-

--------------------

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 DMapper extends Mapper<LongWritable, Text, Text, IntWritable>

{

                private static final int MISSING = 9999;

       

                @Override

                public void map(LongWritable key, Text value, Context context)

                                                throws IOException, InterruptedException

                {

                                if(value.toString().length()>130)

                                {

                                                String line = value.toString();

                                               

                                                String year = line.substring(15, 19);

                                                int airTemperature;

                               

                                                if (line.charAt(87) == '+')

                                                { 

// parseInt doesn't like leading plus

//                                                                                                                                                            // //signs

                                                                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));

                                                }

                                }

                }

} 

DReducer.Java:-

--------------------

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

public class DReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

@Override

public void reduce(Text key, Iterable<IntWritable> values,Context context)

throws IOException, InterruptedException {

int maxValue = Integer.MIN_VALUE;

for (IntWritable value : values) {

maxValue = Math.max(maxValue, value.get());

}

context.write(key, new IntWritable(maxValue));

}

}

Sample Data:

0029029070999991901010106004+64333+023450FM-12+000599999V0202701N015919999999N0000001N9-00781+99999102001ADDGF108991999999999999999999
0029029070999991901010113004+64333+023450FM-12+000599999V0202901N008219999999N0000001N9-00721+99999102001ADDGF104991999999999999999999
0029029070999991901010120004+64333+023450FM-12+000599999V0209991C000019999999N0000001N9-00941+99999102001ADDGF108991999999999999999999
0029029070999991901010206004+64333+023450FM-12+000599999V0201801N008219999999N0000001N9-00611+99999101831ADDGF108991999999999999999999
0029029070999991901010213004+64333+023450FM-12+000599999V0201801N009819999999N0000001N9-00561+99999101761ADDGF108991999999999999999999
0029029070999991901010220004+64333+023450FM-12+000599999V0201801N009819999999N0000001N9-00281+99999101751ADDGF108991999999999999999999
0029029070999991901010306004+64333+023450FM-12+000599999V0202001N009819999999N0000001N9-00671+99999101701ADDGF106991999999999999999999
0029029070999991901010313004+64333+023450FM-12+000599999V0202301N011819999999N0000001N9-00331+99999101741ADDGF108991999999999999999999
0029029070999991901010320004+64333+023450FM-12+000599999V0202301N011819999999N0000001N9-00281+99999101741ADDGF108991999999999999999999
0029029070999991901010406004+64333+023450FM-12+000599999V0209991C000019999999N0000001N9-00331+99999102311ADDGF108991999999999999999999
0029029070999991901010413004+64333+023450FM-12+000599999V0202301N008219999999N0000001N9-00441+99999102261ADDGF108991999999999999999999
0029029070999991901010420004+64333+023450FM-12+000599999V0202001N011819999999N0000001N9-00391+99999102231ADDGF108991999999999999999999
0029029070999991901010506004+64333+023450FM-12+000599999V0202701N004119999999N0000001N9+00001+99999101821ADDGF104991999999999999999999
0029029070999991901010513004+64333+023450FM-12+000599999V0202701N002119999999N0000001N9+00061+99999102591ADDGF104991999999999999999999
0029029070999991901010520004+64333+023450FM-12+000599999V0202301N004119999999N0000001N9+00001+99999102671ADDGF104991999999999999999999