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;

public class LogTime {

	  public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{

	    private Text word = new Text();

	    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
	    	
	    	String line = value.toString();
	    	String[] lineSplit = line.split(" ");
	    	
	    	int time1 = Integer.parseInt(lineSplit[1]);
	    	int time2 = Integer.parseInt(lineSplit[2]);
	    	
	    	int timediff = time2 - time1;
	    	
	    	String ip = lineSplit[0];
	    	word = new Text(ip);
	    	
	    	context.write(word, new IntWritable(timediff));
	    }
	  }

	  public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
	    
		public void reduce(Text key, Iterable<IntWritable> 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 {
	    Configuration conf = new Configuration();
	    Job job = Job.getInstance(conf, "CSVCounter3");
	    job.setJarByClass(CSVCounter3.class);
	    job.setMapperClass(TokenizerMapper.class);
	    job.setCombinerClass(IntSumReducer.class);
	    job.setReducerClass(IntSumReducer.class);
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(IntWritable.class);
	    FileInputFormat.addInputPath(job, new Path(args[0]));
	    FileOutputFormat.setOutputPath(job, new Path(args[1]));
	    System.exit(job.waitForCompletion(true) ? 0 : 1);
	  }
	}