Friday, April 29, 2011

Word Count - Hadoop Map Reduce Example


                Word count is a typical example where Hadoop map reduce developers start their hands on with. This sample map reduce is intended to count the no of occurrences of each word  in the provided input files.

What are the minimum requirements?
1.       Input text files – any text file
2.       Cloudera test VM
3.       The mapper, reducer and driver classes to process the input files

 How it works
                The word count operation takes place in two stages 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
“tring tring the phone rings”
In map phase the sentence would be split as words and form the initial key value pair as
<tring,1>
<tring,1>
<the,1>
<phone,1>
<rings,1>

In the reduce phase the keys are grouped together and the values for similar keys are added. So here there are only one pair of similar keys ‘tring’ the values for these keys would be added so the out put key value pairs would be
<tring,2>
<the,1>
<phone,1>
<rings,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

 
Now coming to the practical side of implementation we need our input file and map reduce program jar to do the process job. In a common map reduce process two methods do the key job namely the map and reduce , the main method would trigger the map and reduce methods. For convenience and readability it is better to include the map , reduce and main methods in 3 different class files . We’d look at the 3 files we require to perform the word count job

Word Count Mapper

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

public class WordCountMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable>
{
      //hadoop supported data types
      private final static IntWritable one = new IntWritable(1);
      private Text word = new Text();
     
      //map method that performs the tokenizer job and framing the initial key value pairs
      public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException
      {
            //taking one line at a time and tokenizing the same
            String line = value.toString();
          StringTokenizer tokenizer = new StringTokenizer(line);
         
          //iterating through all the words available in that line and forming the key value pair
            while (tokenizer.hasMoreTokens())
            {
               word.set(tokenizer.nextToken());
               //sending to output collector which inturn passes the same to reducer
                 output.collect(word, one);
            }
       }
}



Let us dive in details of this source code we can see the usage of a few deprecated classes and interfaces; this is because the code has been written to be compliant with Hadoop versions 0.18 and later. From Hadoop version 0.20 some of the methods are deprecated by still supported.

Lets now focus on the class definition part
implements Mapper<LongWritable, Text, Text, IntWritable>
What does this Mapper<LongWritable, Text, Text, IntWritable> stand for?
The data types provided here are Hadoop specific data types designed for operational efficiency suited for massive parallel and lightning fast read write operations. All these data types are based out of java data types itself, for example LongWritable is the equivalent for long in java, IntWritable for int and Text for String.
When we use it as Mapper<LongWritable, Text, Text, IntWritable> , it refers to the data type of input and output key value pairs specific to the mapper or rateher the map method, ie Mapper<Input Key Type, Input Value Type, Output Key Type, Output Value Type>. In our example the input to a mapper is a single line, so this Text (one input line) forms the input value. The input key would a long value assigned in default based on the position of Text in input file. Our output from the mapper is of the format “Word, 1“ hence the data type of our output key value pair is <Text(String),  IntWritable(int)>

The next key component out here is the map method
map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter)
We’d now look into each of the input parameters in detail. The first and second parameter refers to the Data type of the input Key and Value to the mapper. The third parameter is the output collector which does the job of taking the  output data either from the mapper or reducer, with the output collector we need to specify the Data Types of the output Key and Value from the mapper. The fourth parameter, the reporter is used to report the task status internally in Hadoop environment to avoid time outs.

The functionality of the map method is as follows
1.       Create a IntWritable variable ‘one’ with value as 1
2.       Convert the input line in Text type to a String
3.       Use a tokenizer to split the line into words
4.       Iterate through each word and a form key value pairs as
a.       Assign each work from the tokenizer(of String type) to a Text ‘word
b.      Form key value pairs for each word as <word,one> and push it to the output collector

Word Count Reducer

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

public class WordCountReducer extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable>
{
      //reduce method accepts the Key Value pairs from mappers, do the aggregation based on keys and produce the final out put
      public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException
      {
            int sum = 0;
            /*iterates through all the values available with a key and add them together and give the
            final result as the key and sum of its values*/
          while (values.hasNext())
          {
               sum += values.next().get();
          }
          output.collect(key, new IntWritable(sum));
      }
}


Here like for the mapper the reducer implements
Reducer<Text, IntWritable, Text, IntWritable>
The first two refers to data type of Input Key and Value to the reducer and the last two refers to data type of output key and value. Our mapper emits output as <apple,1> , <grapes,1> , <apple,1> etc. This is the input for reducer so here the data types of key and value in java would be String and int, the equivalent in Hadoop would be Text and IntWritable. Also we get the output as<word, no of occurrences> so the data type of output Key Value would be <Text, IntWritable>

Now the key component here, the reduce method.
The input to reduce method from the mapper after the sort and shuffle phase would be the key with the list of associated values with it. For example here we have multiple values for a single key from our mapper like <apple,1> , <apple,1> , <apple,1> , <apple,1> . This key values would be fed into the reducer as < apple, {1,1,1,1} > .
Now let us evaluate our reduce method
reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter)
Here all the input parameters are hold the same functionality as that of a mapper, the only diference is with the input Key Value. As mentioned earlier the input to a reducer instance is a key and list of values hence  ‘Text key, Iterator<IntWritable> values’ . The next parameter denotes the output collector of the reducer with the data type of output Key and Value.

The functionality of the reduce method is as follows
1.       Initaize a variable ‘sum’ as 0
2.       Iterate through all the values with respect to a key and sum up all of them
3.       Push to the output collector the Key and the obtained sum as value

Driver Class
The last class file is the driver class. This driver class is responsible for triggering the map reduce job in Hadoop, it is in this driver class we provide the name of our job, output key value data types and the mapper and reducer classes. The source code for the same is as follows

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;


public class WordCount extends Configured implements Tool{
      public int run(String[] args) throws Exception
      {
            //creating a JobConf object and assigning a job name for identification purposes
            JobConf conf = new JobConf(getConf(), WordCount.class);
            conf.setJobName("WordCount");

            //Setting configuration object with the Data Type of output Key and Value
            conf.setOutputKeyClass(Text.class);
            conf.setOutputValueClass(IntWritable.class);

            //Providing the mapper and reducer class names
            conf.setMapperClass(WordCountMapper.class);
            conf.setReducerClass(WordCountReducer.class);

            //the hdfs input and output directory to be fetched from the command line
            FileInputFormat.addInputPath(conf, new Path(args[0]));
            FileOutputFormat.setOutputPath(conf, new Path(args[1]));

            JobClient.runJob(conf);
            return 0;
      }
     
      public static void main(String[] args) throws Exception
      {
            int res = ToolRunner.run(new Configuration(), new WordCount(),args);
            System.exit(res);
      }
}


Create all the three java files in your project. Now you’d be having compilation errors just get the latest release of Hadoop and add the jars on to your class path. Once free from compilation errors we have to package them to a jar. If you are using eclipse then right click on the project and use the export utility. While packing  the jar it is better not to give the main class, because in future when you have multiple map reduce and multiple drivers for the same project we should leave an option to choose the main class file  during run time through the command line. 

Follow the steps to execute the job
1.       Copy the jar to a location in LFS (/home/training/usecase/wordcount/wordcount.jar)
2.       Copy the input files from windows to LFS(/home/training/usecase/wordcount/input/)
3.       Create an input directory in HDFS
hadoop fs –mkdir /projects/wordcount/input/
4.       Copy the input files from LFS to HDFS
Hadoop fs –copyFromLocal /home/training/usecase/wordcount/input/* /projects/wordcount/input/
5.       Execute the jar
hadoop jar /home/training/usecase/wordcount/wordcount.jar com.bejoy.samples.wordcount.WordCount /projects/wordcount/input/ /projects/wordcount/output/

We’d just look at the command in detail with each parameter
/home/training/usecase/wordcount/wordcount.jar -> full path of the jar file in LFS
com.bejoy.samples.wordcount.WordCount  -> full package name of the Driver Class
/projects/wordcount/input/  -> input files location in HDFS
/projects/wordcount/output/  -> a directory in HDFS where we need the output files

NOTE: In Hadoop the map reduce process creates the output directory in hdfs and store the output files on to the same. If the output directory already exists in Hadoop then the m/r job wont execute, in that case either you need to change the output directory or delete the provided output directory in HDFS before running the jar again
6.       Once the job shows a success status we can see the output file in the output directory(part-00000)
Hadoop fs –ls /projects/wordcount/output/
7.       For any further investigation of output file we can retrieve the data from hdfs to LFS and from there to the desired location
hadoop fs –copyToLocal /projects/wordcount/output/ /home/training/usecase/wordcount/output/

Some better practices
                In our current example with the configuration parameters or during runtime we are not specifying the number of reducers. In default Hadoop map reduce jobs have the default no of reducers as one, hence one only one reducer instance is used to process the result set from all the mappers and therefore greater the load a single reducer instance and slower the whole process. We are not exploiting parallelism here, to exploit the same we have to assign the no of reducers explicitly. In runtime we can specify the no of reducers as
hadoop jar /home/training/usecase/wordcount/wordcount.jar com.bejoy.samples.wordcount.WordCount -D mapred.reduce.tasks=15 /projects/wordcount/input/ /projects/wordcount/output/

The key point to be noted here is that the no of output files is same as the no of reducers used as every reducer would produce its own output file. All these output files would be available in the hdfs output directory we assigned in the run command. It would be a cumbersome job to combine all these files manually to obtain the result set. For that Hadoop has provided a get merge command

hadoop fs –getmerge /projects/wordcount/output/ /home/training/usecase/wordcount/output/WordCount.txt

This command would combine the contents of all the files available directly within the /projects/wordcount/output/ hdfs directory and write the same to /home/training/usecase/wordcount/output/WordCount.txt file in LFS

You can find the working copy of the word count implementation with hadoop 0.20 API at the following location word count example with hadoop 0.20

Thursday, April 28, 2011

How to run hadoop - map reduce jobs without a cluster? With cloudera VM.

This document is indented to aid basic java developers to kick start practical investigation on Hadoop map reduce jobs without any cluster set up on their end. To understand this document you need to possess basic theoretical knowledge on  Hadoop, hdfs and map reduce jobs. It is also advisable to have some prior knowledge on basic linux commands.
            It is possible to try sample map reduce jobs on your windows PC without any cumbersome hadoop setup if you have cloudera test VM with you. This environment is ideal for functionality testing on very small sample data volumes, larger input data wont be supported due to memory constrains posed by the VM.
To test your sample map reduce job on local hadoop environment (cloudera  training VM) follow the below mentioned steps in order.

1.       Download the following  software and ubuntu image on your windows pc.

2.       Install the VM ware player. Extract the Cloudera Training VM and there you can find a *.vmx file. Open the same.(you can notice that it opens on VM Ware Player)
User credentials for Cloudera VM
User name: training
Password: training

3.       Copy the jar and the required input files into the cloudera linux box.
Here I have copied jar to  home -> training -> use-case -> source-code and
Input files to  home -> training -> use-case -> Input
(To browse to folder in linux box, click on places link on the top menu bar -> select home folder. Now you would be in /home/training folder in your local linux file system from there you can browse and create directories as you do in windows.)
You can copy paste files from your Windows file system to the cloudera linux box just like you do it between folders in Windows.

4.       Open linux terminal(terminal icon available on cloudera linux box desk top)

5.       Create a input directory in your hdfs
Command Syntax:
hadoop fs -mkdir  <full path of directory in hdfs>
Example
hadoop fs -mkdir  /userdata/bejoy/input

6.       Copy the contents from the linux box input folder to hdfs input folder
Command Syntax:
hadoop fs –copyFromLocal  <source directory from local linux box>  <destination directory in hdfs> 
Example
hadoop fs -copyFromLocal  /home/training/use-case/input  /userdata/bejoy/input/

7.       Check the availability of the input files on HDFS(not a mandatory step)
Command Syntax:
hadoop fs –ls  <full path of hdfs directory>
Example
hadoop fs -ls  /userdata/bejoy/input/

8.       Run the jar file with required attributes
Command Syntax:
hadoop jar  <full path of jar with jar name>  <full package name of the Hadoop Driver Class>  < full path of hdfs input directory>  < full path of hdfs output directory>

Example
hadoop jar  /home/training/use-case/source-code/salarycalculator.jar com.ge.corp.hadoop.salcalc.SalaryCalculator  -files  /home/training/use-case/reference/location.txt  /userdata/bejoy/input  /userdata/bejoy/output
Note:  -files option is used if our MR program is referring to other reference files for processing of input files

hadoop jar  /home/training/use-case/source-code/salarycalculator.jar com.ge.corp.hadoop.salcalc.SalaryCalculator  -files  /home/training/use-case/reference/location.txt  -D mapred.reduce.tasks=17  /userdata/bejoy/input  /userdata/bejoy/output
Note:  -D mapred.reduce.tasks is used to specify the no of reducers we’d be using to run our MR job. (the no of output files is same as that of the no of reducers used in running your job)

9.       If you get a log of the below mentioned format in your terminal proceed to step 8 else trouble shoot the error in your MR code and re run the same
10/07/23 03:51:17 INFO mapred.FileInputFormat: Total input paths to process : 2
10/07/23 03:51:18 INFO mapred.JobClient: Running job: job_201007230305_0001
10/07/23 03:51:19 INFO mapred.JobClient:  map 0% reduce 0%
10/07/23 03:51:33 INFO mapred.JobClient:  map 100% reduce 0%
10/07/23 03:51:42 INFO mapred.JobClient:  map 100% reduce 100%
10/07/23 03:51:44 INFO mapred.JobClient: Job complete: job_201007230305_0001
10/07/23 03:51:44 INFO mapred.JobClient: Counters: 18
.
.
.
 Note: this format of log ensures that the MR program ran successfully

10.   Copy the contents of hdfs output directory  to local linux file system
Command Syntax:
hadoop fs –copyFromLocal   <source directory in hdfs>  <destination directory in local linux file system >
Example
hadoop  fs  –copyToLocal  /userdata/bejoy/output  /home/training/use-case/

11.   Open the corresponding directory in your local linux pc and verify the output
(ouput file name would be part-00000.txt)
Here my directory path in local linux pc would be
Home -> training -> use-case -> output

NOTE : Step 10 and 11 is needed only if you need the output file to be put into LFS and the to be transferred to another location or make it accessible foe other applications.
To view the file alone we can use the cat command in hdfs
hadoop fs –cat  <full path of the file in hdfs>

To list the contents of a directory in hdfs we have to use
hadoop fs –ls <full path of the directory in hdfs>