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

import org.apache.hadoop.mapreduce.lib.join.*;

/**
 *
 * @author sc-md
 */
public class SimpleReduceSideJoin
{

    public static abstract class DataMapper
            extends Mapper<Object, Text, IntWritable, Text>
    {
        private IntWritable joinKey = new IntWritable();
        private Text data = new Text();
        private StringBuilder sb = new StringBuilder();

        protected abstract String DataType();

        protected abstract int JoinKeyIndex();

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException
        {
            String line = value.toString();
            String[] parts = line.split(",|\\r|\\n");  // CSV data, split on comma or newline
            
            // JoinKeyIndex represents the position of the joinKey value in the CSV
            // data.
            String productId = parts[JoinKeyIndex()].trim();

            // We have to be able to distinguish between lines of CSV data
            // produced by each input mapper, because the data is combined
            // into one set of partitions before passing to the Reducer.
            // Therefore, we will make the first field of the data the
            // DataType which identifies the kind of data this represents.
            sb.delete(0, sb.length()); // Clear sb
            sb.append(DataType()).append(',');
            sb.append(line);
            data.set(sb.toString());

            joinKey.set(Integer.parseInt(productId));
            context.write(joinKey, data);
        }
    }

    public static class ProductDataMapper
            extends DataMapper
    {
        public static final String DATA_TYPE = "Product";

        @Override
        protected String DataType()
        {
            return DATA_TYPE;
        }

        @Override
        protected int JoinKeyIndex()
        {
            return 0;   // ProductId is at field 0 in CSV data
        }
    }

    public static class ProductSalesDataMapper
            extends DataMapper
    {
        public static final String DATA_TYPE = "Sale";

        @Override
        protected String DataType()
        {
            return DATA_TYPE;
        }

        @Override
        protected int JoinKeyIndex()
        {
            return 0;   // ProductId is at field 0 in CSV data
        }
    }

    public static class JoinReducer
            extends Reducer<IntWritable, Text, NullWritable, Text>
    {
        private StringBuilder sb = new StringBuilder();
        private Text data = new Text();

        @Override
        public void reduce(IntWritable key, Iterable<Text> values,
                Context context) throws IOException, InterruptedException
        {
            String dataType = "";
            String productName = "";
            int totalSales = 0;
            
            // DEBUGGING -- Dump the CSV data lines received with each key
//            data.set("NEW KEY ");
//            context.write(NullWritable.get(), data);
//            for (Text value : values)
//            {
//                sb.delete(0, sb.length());
//                sb.append(key.toString()).append("::");
//                sb.append(value.toString());
//                data.set(sb.toString());
//                context.write(NullWritable.get(), data);
//            }    

            // The "values" represent the lines of CSV data from BOTH the
            // ProductDataMapper and the ProductSalesDataMapper for a specific
            // joinKey (i.e. the ProductId field).
            // Also remember there is only one Product that corresponds to this
            // joinKey i.e. the ProductId is unique to the specific product.
            for (Text value : values)
            {
                String line = value.toString();
                String[] parts = line.split(",|\\r|\\n");  // CSV data, split on comma or newline

                // In the Mapper, we added the DataType as the first
                // field in the data.
                // Product Data Fields: DataType, ProductId, ProductName
                // ProductSales Data Fields: DataType, ProductId, QuantitySold
                dataType = parts[0].trim();
                
                // WARNING: The data in the "values" here is any order, so
                // we cannot assume that the productName will be valid until
                // we exit the for-loop.

                // You only get one Product entry in the "values".
                if (dataType.equals(ProductDataMapper.DATA_TYPE))
                {
                    // Product data
                    // parts[1] is the ProductId
                    productName = parts[2];
                }
                else // if(dataType.equals(ProductSalesDataMapper.DATA_TYPE))   
                {
                    int quantitySold;

                    // BookShop data
                    // parts[1] is the ProductId
                    quantitySold = Integer.parseInt(parts[2].trim());
                    totalSales = totalSales + quantitySold;
                }
            }

            // Output CSV ProductName, TotalSales
            sb.delete(0, sb.length());  // Clear StringBuilder
            sb.append(productName).append(',');
            sb.append(totalSales);

            data.set(sb.toString());

            // The "value" is a line of CSV data.
            // We don't want the results to contain a key followed
            // by a space then the line of CSV. Hence, we NULL the key.
            // 
            context.write(NullWritable.get(), data);

        }
    }

    public static void ExecuteReduceSideJoin(Configuration conf, Job job) throws Exception
    {
        StringBuilder sb = new StringBuilder();

        job.getConfiguration().set("mapreduce.output.basename", "theresults");   // Results in HDFS will be stored as "theresults-r-00000"

        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);   // Reducer produces a NULL key
        job.setOutputValueClass(Text.class);

        job.setReducerClass(SimpleReduceSideJoin.JoinReducer.class);   // NOTE: Just to output the values we could have used setReducerClass(Reducer.class)

        // Order addInputPaths is not important
        MultipleInputs.addInputPath(job, new Path("/mapreducedemo/testdata/products"), TextInputFormat.class, SimpleReduceSideJoin.ProductDataMapper.class);
        MultipleInputs.addInputPath(job, new Path("/mapreducedemo/testdata/productsales"), TextInputFormat.class, SimpleReduceSideJoin.ProductSalesDataMapper.class);
        FileOutputFormat.setOutputPath(job, new Path("/mapreducedemo/testresults/join"));

        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);
        }

        sb.append(System.getProperty("user.dir")).append("/src/mapreduce/productdata.csv");
        fs.copyFromLocalFile(new Path(sb.toString()), new Path("/mapreducedemo/testdata/products/productdata.csv"));   // src, dest    
        sb.delete(0, sb.length()).append(System.getProperty("user.dir")).append("/src/mapreduce/productsalesdata.csv");
        fs.copyFromLocalFile(new Path(sb.toString()), new Path("/mapreducedemo/testdata/productsales/productsalesdata.csv"));   // src, dest     

        // Erase previous results (using HDFS)
        if (fs.exists(new Path("/mapreducedemo/testresults/join")))
        {
            fs.delete(new Path("/mapreducedemo/testresults/join"), true);
        }
    }

    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/join/theresults-r-00000"), new Path(sb.toString()));  // src, dest        
    }
}