/*
 * 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 MapSideJoinNoCache
{
    public static class DataPartitionMapper
       extends Mapper<Object, Text, Text, Text>{

    private Text joinKey = new Text();
    private Text data = new Text();
    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");  // CSV data, split on comma or newline
        int keyIndex = context.getConfiguration().getInt("JoinKeyIndex", 0);
        
        // keyIndex represents the position of the joinKey value in the CSV
        // data.
        
        // Create the new line of data with the joinKey value at the start
        // KeyValueTextInputFormat class assumes that the joinKey is the
        // first value in a line of data.
        if(keyIndex > 0)
        {
            int index = 0;
            
            sb.delete(0, sb.length());  // Clear StringBuilder
            sb.append(parts[keyIndex].trim());
            for (String val : parts)
            {
                if(index != keyIndex)
                {
                    sb.append(',');
                    sb.append(val.trim());
                }
                index++;
            }
            data.set(sb.toString());
        }
        else
        {
            data.set(value.toString());
        }

        // The data is sorted based on the joinKey
        joinKey.set(parts[keyIndex].trim());
        context.write(joinKey, data);
    }
  }
    
    public static class DataPartitionReducer
            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 starting with the joinKey.
                // 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 class JoinMapper
       extends Mapper<Object, TupleWritable, Text, Text>{

    private Text joinKey = new Text();
    private Text joinedData = new Text();
    private StringBuilder sb = new StringBuilder();

    @Override
    public void map(Object key, TupleWritable value, Context context
                    ) throws IOException, InterruptedException {
        
        // CSV data
        // The order in the Tuple, follows the order they were provided in 
        // the CompositeInputFormat.compose() call.
        String bookShopAttributes[] = value.get(0).toString().split(",");
        String bookAttributes[] = value.get(1).toString().split(",");
        
        // NOTE: The joinKey is the first field and has been removed from the 
        // CSV data provided here. Of course, it becomes the "key" argument
        // passed in, so it's not lost.
        // Shop Data Fields: ShopId,ShopName
        // Book Data Fields: BookTitle, BookAuthor
        sb.delete(0, sb.length());  // Clear StringBuilder
        sb.append(key.toString());                  // 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[0].trim());        // Book title
        sb.append(',');
        sb.append(bookAttributes[1].trim());        // Book author

        joinKey.set(key.toString());      // Book Id            
        joinedData.set(sb.toString());
        context.write(joinKey, joinedData);
    }
  }
    
    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);
            }
        }
    }
  
    // REDUCER STAGE IS NOT REQUIRED FOR JoinMapper
  
    public static void ExecuteDataPartitioning(Configuration conf, Job job, String inputFilePath, int joinKeyIndex, String outputPath)throws Exception
    {               
        //job.getConfiguration().set("mapreduce.output.basename", "theresults");   // Results in HDFS will be stored as "theresults-r-00000"
        
        // Store our own value "JoinKeyIndex" in the configuration, so
        // we can access it within the Map task.
        job.getConfiguration().set("JoinKeyIndex", String.valueOf(joinKeyIndex));
        
        //job.getConfiguration().set("mapreduce.task.timeout", "5000");  //ms

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);   // Reducer produces a NULL key
        job.setOutputValueClass(Text.class);
        
        job.setMapperClass(MapSideJoinNoCache.DataPartitionMapper.class);   
        job.setReducerClass(MapSideJoinNoCache.DataPartitionReducer.class);
        
        job.setNumReduceTasks(2);   // Same number of reduce tasks required for each call of this method (just chose 2 here)
        
        FileInputFormat.addInputPath(job, new Path("/mapreducedemo/testdata"));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));  // HDFS path
        
        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(outputPath)))
        {
            fs.delete(new Path(outputPath), true);
        }
        
        StringBuilder sb = new StringBuilder(System.getProperty("user.dir"));
        sb.append(inputFilePath);
        fs.copyFromLocalFile(new Path(sb.toString()), new Path("/mapreducedemo/testdata/testdata.csv"));   // src, dest     
    }
    
    public static void ExecuteMapSideJoin(Configuration conf, Job job, String bookshopDir, String bookDir)throws Exception
    {               
        String joinStatement;
        
        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(MapSideJoinNoCache.JoinMapper.class);
        job.setReducerClass(MapSideJoinNoCache.JoinReducer.class);   // NOTE: Just to output the values we could have used setReducerClass(Reducer.class)
        job.setInputFormatClass(CompositeInputFormat.class);
        
        // Note: KeyValueTextInputFormat assumes that the key starts at the beginning of each line
        // and is followed by a separator byte then the values.
        // The separator byte is specified by mapreduce.input.keyvaluelinerecordreader.key.value.separator.
        joinStatement = CompositeInputFormat.compose("inner",
				KeyValueTextInputFormat.class, 
                                new String[] { bookshopDir, bookDir });
 
        job.getConfiguration().set("mapreduce.join.expr", joinStatement);   // Replaces FileInputFormat.addInputPath
        
        // Assume first field in CSV file is the join key, so KeyValueTextInputFormat should look for
        // first comma.
        job.getConfiguration().set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");
        //job.setNumReduceTasks(0);       // Don't do Reduce stage
        
        FileOutputFormat.setOutputPath(job, new Path("/mapreducedemo/testresults/join"));  
        
        FileSystem fs = FileSystem.get(conf);
        
        // 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        
    }
}