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

import java.io.DataInput;
import java.io.DataOutput;

/**
 *
 * @author sc-md
 */
public class ReduceSideJoin
{
    private static class JoinKey 
            extends CustomKey.Key<String>
    {
        private String key = "";
        
        public JoinKey()
        {
            super(0);   // Initially set sequence number to 0
        }
        
        public JoinKey(String value, int sequenceNumber)
        {
            super(sequenceNumber);
            key = value;
        }
        
        @Override
        public JoinKey setValue(String value)
        {
            key = value;
            return this;
        }

        @Override
        public String getValue()
        {
            return key;
        }

        @Override
        protected String getStringValue()
        {
            return key;
        }

        @Override
        protected boolean valueEqualTo(String value)
        {
            return getValue().equals(value);
        }

        @Override
        protected int valueCompareTo(String value)
        {
            return getValue().compareTo(value);
        }

        @Override
        protected void writeValue(DataOutput out) throws IOException
        {
            out.writeBytes(key);
        }

        @Override
        protected void readValue(DataInput in) throws IOException
        {
            key = in.readLine();
        }
    }
    
    public static class JoinKeyPartitioner extends CustomKey.KeyPartitioner<JoinKey, String>
    {        
    }
    
    public static class JoinKeyGroupComparator extends CustomKey.KeyGroupComparator<String>
    {
        public JoinKeyGroupComparator()
        {
            super(JoinKey.class);
        } 
        
        @Override
        public int compareKeyValues(String v1, String v2)
        {
            return v1.compareTo(v2);
        }
    }
    
    public static abstract class DataMapper
            extends Mapper<Object, Text, JoinKey, Text>
    {
        private JoinKey joinKey = new JoinKey();
        private Text data = new Text();
        private StringBuilder sb = new StringBuilder();

        public abstract int SortOrder();
        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

            // 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
            // sort order number.
            sb.delete(0, sb.length()); // Clear sb
            sb.append(Integer.toString(SortOrder())).append(',');
            sb.append(line);      
            data.set(sb.toString());
            
            // We want the Book data to appear first (see
            // the JoinReducer below), so we use a sort number
            // to ensure this happens.

            joinKey.setSortNumber(SortOrder());                        

            // JoinKeyIndex represents the position of the joinKey value in the CSV
            // data.

            joinKey.setValue(parts[JoinKeyIndex()].trim());
            context.write(joinKey, data);
        }
        
        protected static final int FIRST = 0;
    }    

    public static class BookShopDataMapper
            extends DataMapper
    {
        public static final int SORT_NUMBER = DataMapper.FIRST + 1;  // Shop data follows book data in Map output

        @Override
        public int SortOrder()
        {
            return SORT_NUMBER;
        }

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

    public static class BookDataMapper
            extends DataMapper
    {
        public static final int SORT_NUMBER = DataMapper.FIRST;

        @Override
        public int SortOrder()
        {
            return SORT_NUMBER;
        }

        @Override
        protected int JoinKeyIndex()
        {
            return 0;   // BookId is at field 0 in CSV data
        }
    }
    
    public static class JoinReducer
            extends Reducer<JoinKey, Text, NullWritable, Text>
    {
        private StringBuilder sb = new StringBuilder();
        private Text data = new Text();

        @Override
        public void reduce(JoinKey key, Iterable<Text> values,
                Context context) throws IOException, InterruptedException
        {
            // NOTE: The sort number of the "key" is irrelevant because all the
            // values with the same joinKey value (excluding the sortNumber) are
            // passed together in the list of values.
            String bookTitle = "";
            String bookAuthor = "";
            String shopName = "";
            int sortNumber;

            // 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);
//            }            
            
            // NOTE: We could optimise the following code because
            // the "values" have been sorted so that the first
            // value is from the BookDataMapper and contains the book
            // title and author.
            // Instead of adding the Sort# as an extra data field
            // we could just read the first value from the Iterator
            // like this:
//            Iterator<Text> itr = values.iterator();
//            String[] parts = itr.next().toString().split(",|\\r|\\n");   // Book details
//            bookTitle = parts[1];
//            bookAuthor = parts[2];
//            while (itr.hasNext()) {
//                parts = itr.next().toString().split(",|\\r|\\n");  // Book Shop details
//                shopName = parts[1];
//                // Build the CSV data string
//                context.write(NullWritable.get(), data);
//            }

            // The "values" represent the lines of CSV data from BOTH the
            // BookShopDataMapper and the BookDataMapper for a specific
            // joinKey (i.e. the BookId field).
            // Also remember there is only one Book that corresponds to this
            // joinKey i.e. the BookId is unique to the specific book.
            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 sort number as the first
                // field in the data.
                // Shop Data Fields: Sort#, ShopId, ShopName, BookId
                // Book Data Fields: Sort#, BookId, BookTitle, BookAuthor
                sortNumber = Integer.parseInt(parts[0].trim());

                // You only get one Book entry and because of the sort number
                // it's the first value in the values iterator that we read.
                if (sortNumber == BookDataMapper.SORT_NUMBER)  
                {
                    // Book data
                    // parts[1] is the bookId
                    bookTitle = parts[2];
                    bookAuthor = parts[3];
                }
                else // if(sortNumber == BookShopDataMapper.SORT_NUMBER)   
                {
                    // BookShop data
                    // parts[1] is the shopId
                    shopName = parts[2];
                    // parts[3] is the bookId

                    sb.delete(0, sb.length());  // Clear StringBuilder
                    sb.append(shopName).append(',');
                    sb.append(bookTitle).append(',');
                    sb.append(bookAuthor);

                    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(JoinKey.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);   // Reducer produces a NULL key
        job.setOutputValueClass(Text.class);

        job.setReducerClass(ReduceSideJoin.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/books"), TextInputFormat.class, ReduceSideJoin.BookDataMapper.class);
        MultipleInputs.addInputPath(job, new Path("/mapreducedemo/testdata/shops"), TextInputFormat.class, ReduceSideJoin.BookShopDataMapper.class);
        FileOutputFormat.setOutputPath(job, new Path("/mapreducedemo/testresults/join"));
        
        // We can control the Shuffle and Sort phase that precedes the Reduce process
        // The Partitioner class controls the "shuffle" i.e. which partitions the
        // data is divided into.
        // The GroupingComparator class controls how the data is sorted into the list of
        // "values" passed to the Reducer with the "key".
        job.setPartitionerClass(JoinKeyPartitioner.class);
//        job.setSortComparatorClass(JoinKeySorter.class);   We don't actually need to do this!!
        job.setGroupingComparatorClass(JoinKeyGroupComparator.class);

        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/bookshopdata.csv");
        fs.copyFromLocalFile(new Path(sb.toString()), new Path("/mapreducedemo/testdata/shops/shopdata.csv"));   // src, dest    
        sb.delete(0, sb.length()).append(System.getProperty("user.dir")).append("/src/mapreduce/booksdata.csv");
        fs.copyFromLocalFile(new Path(sb.toString()), new Path("/mapreducedemo/testdata/books/bookdata.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        
    }
}