/*
 * 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.*;

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

    private Text purchaseId = new Text();
    private final static IntWritable one = new IntWritable(1);
    private StringBuilder sb = new StringBuilder();

    @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  "CustomerId, PurchaseId"
        
        // CSV Data is:
        // customerId, productId
        
        // Combine the customerId and productId to create a purchaseId
        // because we want to identify a product has been bought by
        // a specific customer
        sb.delete(0, sb.length());  // Clear StringBuilder
        sb.append(parts[0].trim());
        sb.append(':');
        sb.append(parts[1].trim());
        purchaseId.set(sb.toString());

        context.write(purchaseId, one);
    }
  }
    
  public static class UniqueCustomerPurchaseReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {

    private final static IntWritable one = new IntWritable(1);

    @Override
    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      // Not interested in how many purchases of the same product
      // a customer may have made only that they have purchased it. 
        // So, we can ignore the individual purchases i.e. "values"
      context.write(key, one);   
    }
  }
  
    public static class Stage2DataMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    private Text productId = new Text();
    private final static IntWritable one = new IntWritable(1);

    @Override
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
        String line = value.toString();
        String[] parts = line.split("\\s");   // Split on whitespace (output of previous stage is "customerId:productId   1")
        String purchaseId = parts[0].toString();
        parts = purchaseId.split(":");  // Split on colon to separate customerId and productId from the purchaseId
        
        productId.set(parts[1]);            // Just ProductId required
        context.write(productId, one);
    }
  }
    
  public static class ProductCountReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {


    private IntWritable result = new IntWritable();
    
    @Override
    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);   // ProductId and the quantity of the product
    }
  }
    public static void ExecuteStage1(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(IntWritable.class);
        
        job.setMapperClass(MapReduceChained.Stage1DataMapper.class);   
        //job.setCombinerClass(MapReduceChained.UniqueProductReducer.class);
        job.setReducerClass(MapReduceChained.UniqueCustomerPurchaseReducer.class);
        
        FileInputFormat.addInputPath(job, new Path("/mapreducedemo/testdata"));
        FileOutputFormat.setOutputPath(job, new Path("/mapreducedemo/testresults/stage1"));
        
        FileSystem fs = FileSystem.get(conf);
        
        // Erase previous data and results (using HDFS)
        if(fs.exists(new Path("/mapreducedemo/testdata")))
        {
            fs.delete(new Path("/mapreducedemo/testdata"), 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/customerproductsdata.csv");
        fs.copyFromLocalFile(new Path(sb.toString()), new Path("/mapreducedemo/testdata/testdata.csv"));   // src, dest     
    }
    
    public static void ExecuteStage2(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(IntWritable.class);
        
        job.setMapperClass(MapReduceChained.Stage2DataMapper.class);   
        //job.setCombinerClass(MapReduceChained.ProductCountReducer.class);
        job.setReducerClass(MapReduceChained.ProductCountReducer.class);
        
        FileInputFormat.addInputPath(job, new Path("/mapreducedemo/testresults/stage1"));
        FileOutputFormat.setOutputPath(job, new Path("/mapreducedemo/testresults/stage2"));     
    }
    
    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/stage2/theresults-r-00000"), new Path(sb.toString()));  // src, dest        
    }
}