/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package mapreduce;

import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.conf.*;

//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 mapreduce.WordCount;


/**
 *
 * @author sc-md
 */
public class MapReduce
{ 
    /**
     * @param args the command line arguments
     */
    public static void main(String[] args)throws Exception
    {       
        Job job;
        boolean done = true;
        
        System.out.println("Working Directory = " + System.getProperty("user.dir"));
        
        Configuration conf = new Configuration();
        //conf.set("yarn.resourcemanager.address", "192.168.134.100:8030"); 
        conf.set("yarn.resourcemanager.hostname", "192.168.134.100");
        conf.set("mapreduce.framework.name", "yarn"); 
        conf.set("fs.defaultFS", "hdfs://192.168.134.100:9000"); 
//        conf.set("yarn.application.classpath",        
//                     "$HADOOP_CONF_DIR,$HADOOP_COMMON_HOME/share/hadoop/common/*,$HADOOP_COMMON_HOME/share/hadoop/common/lib/*,"
//                        + "$HADOOP_HDFS_HOME/share/hadoop/hdfs/*,$HADOOP_HDFS_HOME/share/hadoop/hdfs/lib/*,"
//                        + "$HADOOP_YARN_HOME/share/hadoop/yarn/*,$HADOOP_YARN_HOME/share/hadoop/yarn/lib/*,"
//                        + "$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/*,$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/lib/*");
        //conf.set("mapreduce.job.jar",".\\target\\wc-mvn-0.0.1-SNAPSHOT.jar");
        conf.set("mapreduce.app-submission.cross-platform", "true");
        //conf.set("mapreduce.remote.os", "Linux");   // "Windows/Linux"
        conf.set("mapreduce.jobhistory.address", "192.168.134.100:10020");
        //conf.set("mapreduce.map.maxattempts", "1");
        //mapreduce.job.jar 
//        conf.set("mapreduce.appliction.classpath", "$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/*,$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/lib/*");
        
        // Set HADOOP user
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        //conf.set("hadoop.job.ugi", "hadoop");                    // username, comma separated list of groups  

        // Eliminate some warnings that occur during compile
        System.setProperty("hadoop.home.dir", System.getProperty("user.dir")+"\\hadoop-2.8.2");
        System.setProperty("HADOOP_OPTS", "-Djava.library.path="+"F:\\docs\\MikesLectureNotes\\LectureNotes\\AJAX_InteractiveWeb\\BigData\\HDFSapi\\hadoop-2.8.2\\lib");
        System.setProperty("HADOOP_COMMON_LIB_NATIVE_DIR", "F:\\docs\\MikesLectureNotes\\LectureNotes\\AJAX_InteractiveWeb\\BigData\\HDFSapi\\hadoop-2.8.2\\lib\\native");

//        job = CreateJob(conf, "Word Count Job");
//        MapReduceWordCount.Execute(conf, job);     
//        MapReduceCSV.Execute(conf, job);
        
//        job = CreateJob(conf, "Chained Stage1 Job");
//        MapReduceChained.ExecuteStage1(conf, job);
//        job.waitForCompletion(true);
//        job = CreateJob(conf, "Chained Stage2 Job");
//        MapReduceChained.ExecuteStage2(conf, job);
        
//        job = CreateJob(conf, "Book Shop Data Job");
//        MapSideJoinNoCache.ExecuteDataPartitioning(conf, job, "/src/mapreduce/bookshopdata.csv", 2, "/mapreducedemo/joinresults/bookshopdata");
//        if(job.waitForCompletion(true))
//        {
//            job = CreateJob(conf, "Book Data Job");
//            MapSideJoinNoCache.ExecuteDataPartitioning(conf, job, "/src/mapreduce/booksdata.csv", 0, "/mapreducedemo/joinresults/bookdata");
//            if(job.waitForCompletion(true))
//            {
//                job = CreateJob(conf, "Map-Side Join Job");
//                MapSideJoinNoCache.ExecuteMapSideJoin(conf, job, "/mapreducedemo/joinresults/bookshopdata", "/mapreducedemo/joinresults/bookdata");       
//            }
//        }

//        job = CreateJob(conf, "Map-Side Join With Cache Job");
//        MapSideJoinWithCache.ExecuteMapSideJoin(conf, job, "hdfs://192.168.134.100:9000");
        
//        job = CreateJob(conf, "Reduce-Side Join Job");
//        ReduceSideJoin.ExecuteReduceSideJoin(conf, job);
        
//        job = CreateJob(conf, "Reduce-Side Join Job");
//        SimpleReduceSideJoin.ExecuteReduceSideJoin(conf, job);
        
        job = CreateJob(conf, "MP PBSerialisation Job");
        MapReduceWithPBSerialisation.Execute(conf, job);;

        if(done) done = job.waitForCompletion(true);
        if(done)
        {
//          MapReduceWordCount.OnCompletion(conf);
//            MapReduceCSV.OnCompletion(conf);
//            MapReduceChained.OnCompletion(conf);
//            MapSideJoinNoCache.OnCompletion(conf);
//            MapSideJoinWithCache.OnCompletion(conf);
//            ReduceSideJoin.OnCompletion(conf);
//            SimpleReduceSideJoin.OnCompletion(conf);
            MapReduceWithPBSerialisation.OnCompletion(conf);
        }

        System.exit(done ? 0 : 1);   // Return 0 for OK
    }
    
    private static Job CreateJob(Configuration conf, String jobName) throws Exception
    {
        Job job = Job.getInstance(conf, jobName);
        
        // Select the .jar file to load into cluster
        // Either use setJar() or setJarByClass()
        job.setJar(System.getProperty("user.dir")+"\\dist\\MapReduce.jar");       // Use this approach to run through eclipse/netbeans        
        job.setJarByClass(MapReduce.class);   // Use this to run from Command prompt : bin\hadoop -jar MapReduce.jar WordCount       
        
        //job.addCacheFile("file:/" + System.getProperty("user.dir") + "/dist/MapReduce.jar");
        //job.addFileToClassPath(new Path("file:/" + System.getProperty("user.dir") + "/dist/MapReduce.jar"));
        return job;
    }
}