Correction TP4 PLE

TP4.java

package bigdata;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.fs.FileSystem;

public class TP4 {
	public static class TP4Mapper extends Mapper<Object, Text, Text, IntWritable> {
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String tok[] = value.toString().split(",");
			
			if (!tok[4].equals("") && !tok[4].equals("Population")) {
				
				double pop = Math.floor(Math.log10(Double.parseDouble(tok[4])));
				pop = Math.pow(10., pop);
				int pop_class = (int) pop;
				
				context.write(new Text(String.valueOf(pop_class)), new IntWritable(Integer.parseInt(tok[4])));
			}
		}
	}

	public static class TP4Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		
		public void setup(Context context) throws IOException, InterruptedException {
			context.write(new Text("classe\tcount\tavg\tmax\tmin"), null);
		}
		
		public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
			int sum = 0;
			int count = 0;
			int min = Integer.MAX_VALUE, max = Integer.MIN_VALUE;
			for (IntWritable i : values) {
				int current = i.get();
				
				if (current < min)
					min = current;
				if (current > max)
					max = current;
					
				count += current;
				sum++;
			}
			
			StringBuilder b = new StringBuilder();
			b.append(key.toString()).append("\t");
			b.append(String.valueOf(sum)).append("\t");
			b.append(String.valueOf(count/sum)).append("\t");
			b.append(String.valueOf(max)).append("\t");
			b.append(String.valueOf(min)).append("\t");
			
			context.write(new Text(b.toString()), null);
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "TP4-xgerardin");
		
		job.setNumReduceTasks(1);
		job.setJarByClass(TP4.class);
		job.setMapperClass(TP4Mapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setReducerClass(TP4Reducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		job.setInputFormatClass(TextInputFormat.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		if (FileSystem.get(conf).resolvePath(new Path(args[1])) != null)
			FileSystem.get(conf).delete(new Path(args[1]), true);
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}