#mapred

5. Map Reduce Code for Crops Data set.

CropMR.java

package com.example.hadoop; /** * Created by naveen on 9/7/14. */ import java.io.IOException; import java.util.Map; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Partitioner; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.Reducer.Context; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat; import org.apache.hadoop.mapreduce.lib.input.SequenceFileRecordReader; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; public class CropMR { public static class MapDemo extends Mapper<LongWritable, Text, Text, IntWritable> { public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] elements = line.split(","); Text crop = new Text(elements[0]); int quantity = Integer.parseInt(elements[6]); IntWritable it = new IntWritable(quantity); context.write(crop, it); } } public static class ReduceDemo extends Reducer<Text, IntWritable, Text, IntWritable> { // setup, reduce, run, cleanup public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } context.write(key, new IntWritable(sum)); } } public static void main(String[] args) throws Exception { if (args.length < 2) { System.err.println(args.length); System.err.println(args[0]); System.err.println(args[1]); System.err.println("Insufficient args"); System.exit(-1); } Configuration conf = new Configuration(); conf.set("fs.default.name", "hdfs://localhost:9000"); Job job = new Job(conf, "CropDetails"); job.setJarByClass(CropMR.class); // class conmtains mapper and // reducer class job.setMapOutputKeyClass(Text.class); // map output key class job.setMapOutputValueClass(IntWritable.class);// map output value class job.setOutputKeyClass(Text.class); // output key type in reducer job.setOutputValueClass(IntWritable.class);// output value type in // reducer job.setMapperClass(MapDemo.class); job.setReducerClass(ReduceDemo.class); job.setNumReduceTasks(1); job.setInputFormatClass(TextInputFormat.class); // default -- inputkey // type -- longwritable // : valuetype is text job.setOutputFormatClass(TextOutputFormat.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); job.waitForCompletion(true); } }

Now run CropMR.java with arguments /input/crops /output/crops