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


/**
 *
 * @author sc-md
 */
public class MapReduce
{
    public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    
    public void map(Object key, Text value, Reducer.Context context
                    ) throws IOException, InterruptedException {
        String line = value.toString();
      StringTokenizer itr = new StringTokenizer(line);
      Logger.getLogger("TokenizerMapper").debug("Read string " + line );
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }
    
  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values,
                       Reducer.Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }
  
    /**
     * @param args the command line arguments
     */
    public static void main(String[] args)throws Exception
    {
        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.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 job = Job.getInstance(conf, "word count job");
        
        // 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"));
        
        //job.setMapOutputKeyClass(Text.class);               // Specifically set outputs for Mapper class..
        //job.setMapOutputValueClass(IntWritable.class);      // ... alternatively use setOutputKey/Value to set both Mapper and Reducer to same types
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        job.setMapperClass(MapReduce.TokenizerMapper.class);   
        job.setCombinerClass(MapReduce.IntSumReducer.class);
        job.setReducerClass(MapReduce.IntSumReducer.class);
        
//        job.setInputFormatClass(TextInputFormat.class);
//        job.setOutputFormatClass(TextOutputFormat.class);

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

        // Erase previous results
        FileSystem.get(conf).delete(new Path("/mapreducedemo/results2"), true);
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}