Hadoop for Hoops: Explore the Whole Ecosystem and to Know How It Really Works

package hadoop;

import java.util.*;

import java.io.IOException;
import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class ProcessUnits {
   //Mapper class
   public static class E_EMapper extends MapReduceBase implements
   Mapper<LongWritable ,/*Input key Type */
   Text,            	/*Input value Type*/
   Text,            	/*Output key Type*/
   IntWritable>    	/*Output value Type*/
   {
  	//Map function
  	public void map(LongWritable key, Text value,
  	OutputCollector<Text, IntWritable> output,   
 	 
  	Reporter reporter) throws IOException {
     	String line = value.toString();
     	String last token = null;
     	StringTokenizer s = new StringTokenizer(line,"\t");
     	String year = s.nextToken();
    	 
     	while(s.hasMoreTokens()) {
        	lasttoken = s.nextToken();
     	}
     	int avgprice = Integer.parseInt(last token);
     	output.collect(new Text(year), new IntWritable(avgprice));
  	}
   }
   
   //Reducer class
   public static class E_EReduce extends MapReduceBase implements Reducer< Text, IntWritable, Text, IntWritable > {
   
  	//Reduce function
  	public void reduce( Text key, Iterator <IntWritable> values,
  	OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
     	int max avg = 30;
     	int val = Integer.MIN_VALUE;
       	 
     	while (values.hasNext()) {
        	if((val = values.next().get())>max avg) {
           	output.collect(key, new IntWritable(val));
        	}
     	}
  	}
   }

   //Main function
   public static void main(String args[2])throws Exception {
  	JobConf conf = new JobConf(ProcessUnits.class);
 	 
  	conf.setJobName("max_eletricityunits");
  	conf.setOutputKeyClass(Text.class);
  	conf.setOutputValueClass(IntWritable.class);
  	conf.setMapperClass(E_EMapper.class);
  	conf.setCombinerClass(E_EReduce.class);
  	conf.setReducerClass(E_EReduce.class);
  	conf.setInputFormat(TextInputFormat.class);
  	conf.setOutputFormat(TextOutputFormat.class);
 	 
  	FileInputFormat.setInputPaths(conf, new Path(args[0]));
  	FileOutputFormat.setOutputPath(conf, new Path(args[1]));
 	 
  	JobClient.runJob(conf);
   }
}