Big Data Analytics: Running a Wordcount Mapreduce example in Hadoop 2.7

Running a Wordcount Mapreduce example in Hadoop 2.7

Here I have explained how to run a MapReduce word count example on a Hadoop cluster in CentOS. The program is written in Java. It counts word occurrences in a text file, and then outputs a new text file that contains each word paired with its frequency of occurrence.

Pre-requisites:

· Eclipse

· Hadoop 2.x

· Java

How it works:

The word count operation takes place in two stages. They are a mapper phase and a reducer phase. In mapper phase, first the test is tokenized into words then we form a key value pair with these words where the key being the word itself and value ‘1’.

For example consider the sentence

“Where in the world”

“What in the world”

In map phase the sentence would be split as words and form the initial key value pair as

<Where,1>

<in,1>

<the,1>

<world,1>

<what,1>

<in,1>

<the,1>

<world,1>

In the reduce phase, the keys are grouped together and the values for similar keys are added. So here there are few pairs of similar keys “in”, “the”, “world” and there values would be added.

The output key value pairs would be

<Where,1>

<in,2>

<the,2>

<world,2>

<what,1>

This would give the number of occurrence of each word in the input. Thus reduce forms an aggregation phase for keys.

The point to be noted here is that first the mapper class executes completely on the entire data set splitting the words and forming the initial key value pairs. Only after this entire process is completed the reducer starts. Say if we have a total of 10 lines in our input files combined together, first the 10 lines are tokenized and key value pairs are formed in parallel, only after this the aggregation/ reducer would start its operation.

The figure below would throw more light to your understanding

Input Files

Each line passed to mapper instances

Map key value splitting

Sort & shuffle

Reduce key value pairs

Final output

This program helps us in getting a good understanding of parallel processing of hadoop.

It consists of three classes.

· Driver class- which is the main class

· Mapper class- which does the map functions

· Reducer class- which does the reduce functions

Driver Class

package com.pravat.hadoop;

import java.io.IOException;

import java.util.Date;

import java.util.Formatter;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import org.apache.hadoop.util.GenericOptionsParser;

public class WordCountDriver {

public static void main(String[] args) throws IOException,

InterruptedException, ClassNotFoundException {

Configuration conf = new Configuration();

GenericOptionsParser parser = new GenericOptionsParser(conf, args);

args = parser.getRemainingArgs();

Job job = new Job(conf, "wordcount");

job.setJarByClass(WordCountDriver.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

job.setInputFormatClass(TextInputFormat.class);

job.setOutputFormatClass(TextOutputFormat.class);

Formatter formatter = new Formatter();

FileInputFormat.setInputPaths(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job, new Path(args[1]));

job.setMapperClass(WordCountMapper.class);

job.setReducerClass(WordCountReducer.class);

System.out.println(job.waitForCompletion(true));

}

Mapper Class

package com.pravat.hadoop;

import java.io.IOException;

import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

private Text word = new Text();

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

protected void map(LongWritable key, Text value, Context context)

throws IOException, InterruptedException {

String line = value.toString();

StringTokenizer tokenizer = new StringTokenizer(line);

while (tokenizer.hasMoreTokens()) {

word.set(tokenizer.nextToken());

context.write(word, one);

}

Reducer Class

package com.pravat.hadoop;

import java.io.IOException;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

public class WordCountReducer extends

Reducer<Text, IntWritable, Text, IntWritable> {

protected void reduce(Text key, Iterable<IntWritable> values,

Context context) throws IOException, InterruptedException {

int sum = 0;

for (IntWritable value : values) {

sum += value.get();

}

context.write(key, new IntWritable(sum));

}

The jars necessary for this code is taken from the same version of hadoop package which is installed in the cluster. If the version is different, then it will result in error.
Here the mapper class reads the input file line by line. Then inside the mapper class, we convert the line to string after that we tokenize it into words. ie each line is split into individual words. The output of the mapper class is given to the reducer. the output of the mapper is in the form of a pair.
The context.write method actually gives a key-value pair to the reducer. Here the key is the word and value is “one” which is a variable assigned to the value 1.

In the Reducer, we merges these words and counts the values attached to similar words.
For example if we give an input file as