Hands on tour of PySpark

The code is available at
https://github.com/PravatSutar/SparkExamples

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

Bigdata Hadoop - Cluster Management through CLUSH

Clush is an open source package that lets you use a parallel shell to simplify cluster management from the command line and shell. This topic covers how to install clush and run a utility that does cluster auditing for you. This step is extremely useful before you deploy your Hadoop cluster.

a.       Clush only needs to be installed on one node, e.g. the edge node or first node in the cluster. Clustershell is a package available for CentOS, Redhat and Ubuntu. e.g.

#yum --enablerepo=epel install clustershell

b.      Prerequisite: passwordless ssh to all nodes in the cluster for root or an account with sudo rights.

c.       Define at least one group (all) in /etc/clustershell/groups. e.g. all: node[1-9]

In our dev clusters, it is

#cd /etc/clustershell

#cat groups

all: node1.pravatsutar.com node2.pravatsutar.com node3.pravatsutar.com

d.      Test install/config with:

#clush -a date

Compare clush output with and without -b option:

#clush -a date and #clush -ab date

Clush Commands:

To copy to all the nodes

clush -a -c /home/clduser/mprrepo_bkp/jdk-7u51-linux-x64.rpm

To remove the files from the nodes

clush -g all rm -rf /home/clduser/mprrepo_bkp/jdk-7u51-linux-x64.rpm

If you want to install CLDB

clush -g cldb <mapr cldb rpm>

Use below command if you have any manual intervention to press YES

clush -g all yum -y install ganglia-gmond

To copy from one to another directory

clush -g all -c /etc/ganglia/gmond.conf --dest /etc/ganglia/

Uninstall all the rpm files from all the nodes

clush -g all yum -y remove mapr-core-3.1.0.23703.GA-1.x86_64

Command To erase the java

Clush –g all yum –y erase java

Command To delete the files

Clush –g all rm /test/test.txt

To append teradatahost names to our /etc/hosts file

clush -g all cat '/etc/terahosts >> /etc/hosts'

Where the Extract Files wants to place:
clush -g all tar -xf /usr/lib64/python2.6/numpy-1.9.1.tar.gz -C /usr/lib64/python2.6/

To Remove JAVA

Clush –g all -yum -y erase jdk

clush -g all yum -y install /home/clduser/jdk-7u51-linux-x64.rpm