@@ -0,0 +1,2139 @@
+import java.io.IOException;%0Aimport java.util.StringTokenizer;%0A%0Aimport org.apache.hadoop.conf.Configuration;%0Aimport org.apache.hadoop.fs.Path;%0Aimport org.apache.hadoop.io.IntWritable;%0Aimport org.apache.hadoop.io.Text;%0Aimport org.apache.hadoop.mapreduce.Job;%0Aimport org.apache.hadoop.mapreduce.Mapper;%0Aimport org.apache.hadoop.mapreduce.Reducer;%0Aimport org.apache.hadoop.mapreduce.lib.input.FileInputFormat;%0Aimport org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;%0A%0Apublic class BirdWeightSum %7B%0A%0A%09public static class BirdWeightMapper extends%0A%09%09%09Mapper%3CObject, Text, Text, IntWritable%3E %7B%0A%0A%09%09private final static IntWritable weight = new IntWritable(1);%0A%09%09private Text timeAndStation = new Text();%0A%09%09private Log log;%0A%0A%09%09public void map(Object key, Text value, Context context)%0A%09%09%09%09throws IOException, InterruptedException %7B%0A%09%09%09StringTokenizer itr = new StringTokenizer(value.toString());%0A%09%09%09while (itr.hasMoreTokens()) %7B%0A%09%09%09%09log = new Log(itr.nextToken());%0A%09%09%09%09timeAndStation.set(log.getTowerId() + %22,%22 + log.getDate());%0A%09%09%09%09weight.set(Integer.parseInt(log.getWeight()));%0A%09%09%09%09context.write(timeAndStation, weight);%0A%09%09%09%7D%0A%09%09%7D%0A%09%7D%0A%0A%09public static class BirdWeightReducer extends%0A%09%09%09Reducer%3CText, IntWritable, Text, IntWritable%3E %7B%0A%0A%09%09private IntWritable result = new IntWritable();%0A%0A%09%09public void reduce(Text key, Iterable%3CIntWritable%3E values,%0A%09%09%09%09Context context) throws IOException, InterruptedException %7B%0A%09%09%09int sum = 0;%0A%09%09%09for (IntWritable val : values) %7B%0A%09%09%09%09sum += val.get();%0A%09%09%09%7D%0A%09%09%09result.set(sum);%0A%09%09%09context.write(key, result);%0A%09%09%7D%0A%09%7D%0A%0A%09public static void main(String%5B%5D args) throws Exception %7B%0A%09%09Configuration conf = new Configuration();%0A%09%09Job job = Job.getInstance(conf, %22bird weight sum%22);%0A%09%09job.setJarByClass(BirdWeightSum.class);%0A%09%09job.setMapperClass(BirdWeightMapper.class);%0A%09%09job.setCombinerClass(BirdWeightReducer.class);%0A%09%09job.setReducerClass(BirdWeightReducer.class);%0A%09%09job.setOutputKeyClass(Text.class);%0A%09%09job.setOutputValueClass(IntWritable.class);%0A%09%09FileInputFormat.addInputPath(job, new Path(args%5B0%5D));%0A%09%09FileOutputFormat.setOutputPath(job, new Path(args%5B1%5D));%0A%09%09System.exit(job.waitForCompletion(true) ? 0 : 1);%0A%09%7D%0A%7D

Add map reduce function to calculate sum of weights