/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package mapreduce;

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

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.mapreduce.*;
//import org.apache.hadoop.util.*;
import org.apache.log4j.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;

/**
 *
 * @author sc-md
 */
public class MapReduceCSV
{
    public static class CSVDataMapper
       extends Mapper<Object, Text, Text, DoubleWritable>{

    private DoubleWritable yearAvg = new DoubleWritable();
    private Text name = new Text();

    @Override
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
        String line = value.toString();
        String[] parts = line.split(",|\\r|\\n");  // Split on comma or newline
        
        // CSV Data is:
        // year, name, value1, value2, avg
        
        // Key is the name of the person which is found in the second column 
        // and the value is the year average found in the last column
        name.set(parts[1].trim());
        yearAvg.set(Double.parseDouble(parts[4]));   // parse also trims spaces
        
        context.write(name, yearAvg);
    }
  }
    
  public static class BestYearAvgReducer
       extends Reducer<Text,DoubleWritable,Text,DoubleWritable> {
    private DoubleWritable bestOverall = new DoubleWritable();

    @Override
    public void reduce(Text key, Iterable<DoubleWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      double bestAvg = 0;
      for (DoubleWritable val : values) {
          if(val.get() > bestAvg)
          {
              bestAvg = val.get();  // Find the best year average for the person
          }
      }
      bestOverall.set(bestAvg);
      context.write(key, bestOverall);   // The key is the name of the person
    }
  }
  
    public static void Execute(Configuration conf, Job job)throws Exception
    {               
        job.getConfiguration().set("mapreduce.output.basename", "theresults");   // Results in HDFS will be stored as "theresults-r-00000"
                
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        
        job.setMapperClass(MapReduceCSV.CSVDataMapper.class);   
        job.setCombinerClass(MapReduceCSV.BestYearAvgReducer.class);
        job.setReducerClass(MapReduceCSV.BestYearAvgReducer.class);
        
        FileInputFormat.addInputPath(job, new Path("/mapreducedemo/csvfiles"));
        FileOutputFormat.setOutputPath(job, new Path("/mapreducedemo/testresults"));
        

//        String csvData = "2009,Fred,22,8,15\n" +
//                        "2009,Bob,3,55,29\n" +
//                        "2010,Fred,44,884,464\n" +
//                        "2010,Bob,555,354,454.5\n" +
//                        "2010,Jill,3,54,28.5\n" +
//                        "2011,Sue,55,63,59\n" +
//                        "2012,Bob,37,95,66\n" +
//                        "2012,Jill,24,133,78.5";
        FileSystem fs = FileSystem.get(conf);
        
        // Erase previous data and results (using HDFS)
        if(fs.exists(new Path("/mapreducedemo/csvfiles")))
        {
            fs.delete(new Path("/mapreducedemo/csvfiles"), true);
        }
                
        if(fs.exists(new Path("/mapreducedemo/testresults")))
        {
            fs.delete(new Path("/mapreducedemo/testresults"), true);
        }
        
        StringBuilder sb = new StringBuilder(System.getProperty("user.dir"));
        sb.append("/src/mapreduce/testdata.csv");
        fs.copyFromLocalFile(new Path(sb.toString()), new Path("/mapreducedemo/csvfiles/testdata.csv"));   // src, dest
               
        //sb.delete(0, sb.length());  // Clear StringBuilder       
    }
    
    public static void OnCompletion(Configuration conf) throws Exception
    {
        FileSystem fs = FileSystem.get(conf);
        StringBuilder sb = new StringBuilder(System.getProperty("user.dir"));
        sb.append("/src/mapreduce/results");
        fs.copyToLocalFile(new Path("/mapreducedemo/testresults/theresults-r-00000"), new Path(sb.toString()));  // src, dest        
    }
}