// Defining package of the class
package com.PACKT.chapter1;

// Importing java libraries 
import java.io.*;
importjava.util.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

// Defining the Map class
public class Map extends MapReduceBase implements
         Mapper<LongWritable, 
                Text, 
                Text, 
                IntWritable>{

//Defining the map method – for processing the data with // problem specific logic
public void map(LongWritable key,
                Text value,
                OutputCollector<Text,
                IntWritable> output,
                Reporter reporter) 
                throws IOException {

// For breaking the string to tokens and convert them to lowercase
StringTokenizer st = new StringTokenizer(value.toString().toLowerCase());

// For every string tokens
while(st.hasMoreTokens()) {

// Emitting the (key,value) pair with value 1.
output.collect(new Text(st.nextToken()), 
               new IntWritable(1));
        }

    }

}

// Defining package of the class
package com.PACKT.chapter1;

// Importing java libraries
import java.io.*;
importjava.util.*;
import org.apache.hadoop.io.*;
importorg.apache.hadoop.mapred.*;

// Defining the Reduce class 
public class Reduce extends MapReduceBase implements
          Reducer<Text,
                  IntWritable,
                  Text,
                  IntWritable> {

// Defining the reduce method for aggregating the //generated output of Map phase
public void reduce(Text key,
                   Iterator<IntWritable> values,
                   OutputCollector<Text,IntWritable>
                   output, 
                   Reporter reporter) throws IOException {

// Setting initial counter value as 0
int count = 0;

// For every element with similar key attribute, increment its counter value by adding 1.
while(values.hasNext()) {
count += values.next().get();
        }

// Emitting the (key,value) pair
output.collect(key, new IntWritable(count));
    }
}

//Defining package of the class
package com.PACKT.chapter1;

// Importing java libraries
import java.io.*;
importorg.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;
importorg.apache.hadoop.mapred.*;
importorg.apache.hadoop.util.*;
importorg.apache.hadoop.conf.*;

//Defining wordcount class for job configuration 
  // information
public class WordCount extends Configured implements Tool{

publicint run(String[] args) throws IOException{
JobConfconf = new JobConf(WordCount.class);
conf.setJobName("wordcount");

//For defining the output key format
conf.setOutputKeyClass(Text.class);

//For defining the output value format
conf.setOutputValueClass(IntWritable.class);

// For defining the Mapper class implementation
conf.setMapperClass(Map.class);

// For defining the Reducer class implementation
conf.setReducerClass(Reduce.class);

// For defining the type of input format 
conf.setInputFormat(TextInputFormat.class);

// For defining the type of output format
conf.setOutputFormat(TextOutputFormat.class);

// For defining the command line argument sequence for // input dataset path
FileInputFormat.setInputPaths(conf, new Path(args[0]));

// For defining the command line argument sequence for // output dataset path
FileOutputFormat.setOutputPath(conf, new Path(args[1]));


// For submitting the configuration object
JobClient.runJob(conf);

return 0;
    }

// Defining the main() method to start the execution of // the MapReduce program
public static void main(String[] args) throws Exception {
  intexitCode = ToolRunner.run(new WordCount(), args);
  System.exit(exitCode); } }