/*
 * 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.net.URI;
import java.io.BufferedReader;
import java.io.FileReader;

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

    public static class JoinMapper
            extends Mapper<Object, Text, Text, Text>
    {

        private Map<String, String> bookList = new HashMap<>();
        private Text joinKey = new Text();
        private Text data = new Text();
        private StringBuilder sb = new StringBuilder();

        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException
        {
            loadBookListInToMemory(context);
        }

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException
        {
            int keyIndex = 2;  // joinKey field position in CSV book shop data (i.e. bookId)
            String line = value.toString();
            String[] bookShopAttributes = line.split(",|\\r|\\n");  // CSV data, split on comma or newline
            String joinKeyValue = bookShopAttributes[keyIndex].trim();
            String bookAttributes[] = bookList.get(joinKeyValue).split(",|\\r|\\n");
            
            // Shop Data Fields: ShopId, ShopName, BookId
            // Book Data Fields: BookId, BookTitle, BookAuthor
            
            //bookShopAttributes[2]  // Book Id at index 2
            //bookAttributes[0]      // Book Id at index 0
            
            sb.delete(0, sb.length());  // Clear StringBuilder
            sb.append(joinKeyValue);                  // Book Id
            sb.append(',');
            sb.append(bookShopAttributes[0].trim());    // Shop Id
            sb.append(',');
            sb.append(bookShopAttributes[1].trim());    // Shop name
            sb.append(',');
           
            
            sb.append(bookAttributes[1].trim());        // Book title
            sb.append(',');
            sb.append(bookAttributes[2].trim());        // Book author

            // keyIndex represents the position of the joinKey value in the CSV
            // data.       
            joinKey.set(joinKeyValue);
            data.set(sb.toString());
            context.write(joinKey, data);
        }

        private void loadBookListInToMemory(Context context)
        {
            int keyIndex = 0;  // joinKey field (bookId) position in CSV book data
            
            // Read the "books.csv" from the distributed cache
            try (BufferedReader br = new BufferedReader(new FileReader("bookdata.csv")))
            {
                String line;
                while ((line = br.readLine()) != null)
                {
                    String columns[] = line.split(",|\\r|\\n");
                    bookList.put(columns[keyIndex].trim(), line);
                }
            }
            catch (IOException e)
            {
                e.printStackTrace();
            }
        }
    }

    public static class JoinReducer
            extends Reducer<Text, Text, NullWritable, Text>
    {

        @Override
        public void reduce(Text key, Iterable<Text> values,
                Context context) throws IOException, InterruptedException
        {
            for (Text value : values)
            {
                // 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(), value);
            }
        }
    }

    public static void ExecuteMapSideJoin(Configuration conf, Job job, String hdfsURI) 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(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);   // Reducer produces a NULL key
        job.setOutputValueClass(Text.class);

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

        FileInputFormat.addInputPath(job, new Path("/mapreducedemo/testdata/shops"));
        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/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);
        }
        sb.delete(0, sb.length()).append(hdfsURI).append("/mapreducedemo/testdata/books/bookdata.csv");
        job.addCacheFile(new URI(sb.toString()));
    }

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