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

import com.twitter.elephantbird.mapreduce.input.*;
import com.twitter.elephantbird.mapreduce.output.*;
import com.twitter.elephantbird.mapreduce.io.*;
import com.twitter.elephantbird.util.*;

import mapreduce.proto.ProtocolBuffersTest.Age;

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

    public static class ProtocolBufferMapper
            extends Mapper<Object, Text, Text, ProtobufWritable<Age>>
    {
        Text nameKey = new Text();
        ProtobufWritable<Age> protoWritable = ProtobufWritable.newInstance(Age.class);        

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException
        {
            String line = value.toString();
            String[] parts = line.split(",|\\r|\\n");
            String name = parts[0].trim();
            int age = Integer.parseInt(parts[1].trim());

            protoWritable.set(Age.newBuilder()
                    .setName(name)
                    .setAge(age)
                    .build());
            
            nameKey.set(name);
            context.write(nameKey, protoWritable);
        }
    }

    public static class ProtocolBufferReducer
            extends Reducer<Text, ProtobufWritable<Age>, NullWritable, Text>
    {
        ProtobufConverter<Age> converter = ProtobufConverter.newInstance(Age.class);
        private StringBuilder sb = new StringBuilder();
        private Text data = new Text();

        @Override
        protected void reduce(Text key, Iterable<ProtobufWritable<Age>> values, Context context)
                throws IOException, InterruptedException
        {
            String name;
            int age;
            
            for (ProtobufWritable<Age> value : values)
            {
                /* setConverter() before value.get() is invoked. It is required since
                 * 'value' object was created by MR with ProtobufWritable's default
                 * constructor, as result object does not know its runtime Protobuf class.
                 * The 'value' object just contains serialized bytes for the protobuf.
                 * It can create the actual protobuf only after it knows which protobuf
                 * class to use.
                 */
                value.setConverter(converter);
                name = value.get().getName();
                age = value.get().getAge();
                
                sb.delete(0, sb.length());  // Clear StringBuilder
                sb.append(name).append(',');
                sb.append(age);

                data.set(sb.toString());
                
                context.write(NullWritable.get(), data);
            }
        }
    }

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

        job.getConfiguration().set("mapreduce.output.basename", "theresults");
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(ProtobufWritable.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        job.setMapperClass(MapReduceWithPBSerialisation.ProtocolBufferMapper.class);
        //job.setCombinerClass(MapReduceWithPBSerialisation.ProtocolBufferReducer.class);
        job.setReducerClass(MapReduceWithPBSerialisation.ProtocolBufferReducer.class);

//        job.setInputFormatClass(TextInputFormat.class);
//        job.setOutputFormatClass(TextOutputFormat.class);

        //job.setUser("hadoop");
        FileInputFormat.addInputPath(job, new Path("/mapreducedemo/testdata"));
        FileOutputFormat.setOutputPath(job, new Path("/mapreducedemo/testresults"));

        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/agedata.csv");
        fs.copyFromLocalFile(new Path(sb.toString()), new Path("/mapreducedemo/testdata/agedata.csv"));   // src, dest    

        // Erase previous results (using HDFS)
        if (fs.exists(new Path("/mapreducedemo/testresults")))
        {
            fs.delete(new Path("/mapreducedemo/testresults"), 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/theresults-r-00000"), new Path(sb.toString()));  // src, dest    
    }
}