WordCountPartitioner.
java
Objective:
Write a MapReduce program to count the occurrence of similar words in a file.
Use a partitioner to partition key/value pairs based on alphabets.
Partitioner class ([Link])
import [Link];
import [Link];
import [Link];
public class WordCountPartitioner extends Partitioner<Text, IntWritable> {
@Override
public int getPartition(Text key, IntWritable value, int numPartitions) {
String word = [Link]();
char alphabet = [Link]().charAt(0);
int partitionNumber = 0;
switch (alphabet) {
case 'A': partitionNumber = 1; break;
case 'B': partitionNumber = 2; break;
case 'C': partitionNumber = 3; break;
case 'D': partitionNumber = 4; break;
case 'E': partitionNumber = 5; break;
case 'F': partitionNumber = 6; break;
case 'G': partitionNumber = 7; break;
case 'H': partitionNumber = 8; break;
case 'I': partitionNumber = 9; break;
case 'J': partitionNumber = 10; break;
case 'K': partitionNumber = 11; break;
case 'L': partitionNumber = 12; break;
case 'M': partitionNumber = 13; break;
case 'N': partitionNumber = 14; break;
case 'O': partitionNumber = 15; break;
case 'P': partitionNumber = 16; break;
case 'Q': partitionNumber = 17; break;
case 'R': partitionNumber = 18; break;
case 'S': partitionNumber = 19; break;
case 'T': partitionNumber = 20; break;
case 'U': partitionNumber = 21; break;
case 'V': partitionNumber = 22; break;
case 'W': partitionNumber = 23; break;
case 'X': partitionNumber = 24; break;
case 'Y': partitionNumber = 25; break;
case 'Z': partitionNumber = 26; break;
1
Adithya M (Dept. of CSE, CEC)
default: partitionNumber = 0; break;
}
// Hadoop expects partition number to be in range [0, numPartitions-1]
return partitionNumber % numPartitions;
}
}
Driver Class ([Link])
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public class WordCountDriver {
public static void main(String[] args) throws Exception {
if ([Link] != 2) {
[Link]("Usage: WordCountDriver <input path> <output path>");
[Link](-1);
}
Configuration conf = new Configuration();
Job job = [Link](conf, "Word Count with Custom Partitioner");
[Link]([Link]);
// Set Mapper and Reducer classes (assumed to exist)
[Link]([Link]);
[Link]([Link]);
// Set output key and value types
[Link]([Link]);
[Link]([Link]);
// Set Partitioner class
[Link]([Link]);
// Set number of reduce tasks (must match partitions)
[Link](27); // A–Z + default
// Set input and output paths
[Link](job, new Path(args[0]));
2
Adithya M (Dept. of CSE, CEC)
[Link](job, new Path(args[1]));
// Submit job and wait for completion
[Link]([Link](true) ? 0 : 1);
}
}
Input Data
A text file containing lines such as:
• Welcome to Hadoop Session
• Introduction to Hadoop
• Introducing Hive
• Hive Session
• Pig Session
MapReduce Searching
Input file (student records) is stored in HDFS.
1001,John,45
1002,Jack,39
1003,Alex,44
1004,Smith,38
1005,Bob,33
Driver ([Link]) sets the job configuration and specifies the keyword to search
(e.g., "Jack").
Mapper ([Link]) reads line by line and checks if the keyword is present.
• If yes, it emits (line, position) as (key, value).
Reducer ([Link]) simply outputs the key-value pairs received from the
mapper.
Final output is stored in HDFS output path with only the lines that matched the keyword.
1. Mapper ([Link])
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public class WordSearchMapper extends Mapper<LongWritable, Text, Text, Text> {
static String keyword;
static int pos = 0;
3
Adithya M (Dept. of CSE, CEC)
protected void setup(Context context) throws IOException, InterruptedException {
Configuration configuration = [Link]();
keyword = [Link]("keyword");
}
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
InputSplit i = [Link](); // Get the input split for this map
FileSplit f = (FileSplit) i;
String fileName = [Link]().getName();
Integer wordPos;
pos++;
if ([Link]().contains(keyword)) {
wordPos = [Link](keyword);
[Link](value, new Text(fileName + "," + new IntWritable(pos).toString()
+ "," + [Link]()));
}
}
}
2. Reducer ([Link])
import [Link];
import [Link];
import [Link];
public class WordSearchReducer extends Reducer<Text, Text, Text, Text> {
protected void reduce(Text key, Iterable<Text> value, Context context)
throws IOException, InterruptedException {
for (Text val : value) {
[Link](key, val);
}
}
}
3. Driver ([Link])
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public class WordSearcher {
4
Adithya M (Dept. of CSE, CEC)
public static void main(String[] args) throws IOException, InterruptedException,
ClassNotFoundException {
Configuration conf = new Configuration();
[Link]("keyword", "Jack"); // keyword to search
Job job = [Link](conf, "Word Search");
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
[Link](1);
[Link](job, new Path("/mapreduce/[Link]"));
[Link](job, new Path("/mapreduce/output/search"));
[Link]([Link](true) ? 0 : 1);
}
}
MapReduce Sorting
Objective & Input Data
• Objective:
To write a MapReduce program that sorts student records by student name.
• Input Data ([Link]):
1001,John,45
1002,Jack,39
1003,Alex,44
1004,Smith,38
1005,Bob,33
Expected Sorted Output (by name):
1003,Alex,44
1005,Bob,33
1002,Jack,39
1001,John,45
1004,Smith,38
1. Mapper (SortMapper inside [Link])
import [Link];
import [Link];
5
Adithya M (Dept. of CSE, CEC)
import [Link];
import [Link];
public class SortStudNames {
public static class SortMapper extends Mapper<LongWritable, Text, Text, Text> {
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] token = [Link]().split(",");
// token[0] = ID, token[1] = Name, token[2] = Marks
[Link](new Text(token[1]), new Text(token[0] + "," + token[2]));
}
}
2. Reducer (SortReducer inside [Link])
import [Link];
import [Link];
import [Link];
import [Link];
public static class SortReducer extends Reducer<Text, Text, NullWritable, Text> {
public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
for (Text details : values) {
[Link]([Link](), details);
}
}
}
3. Driver (main method inside [Link])
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public static void main(String[] args) throws IOException, InterruptedException,
ClassNotFoundException {
Configuration conf = new Configuration();
Job job = [Link](conf, "Sort Students by Name");
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
6
Adithya M (Dept. of CSE, CEC)
[Link]([Link]);
[Link](job, new Path("/mapreduce/[Link]"));
[Link](job, new Path("/mapreduce/output/sorted"));
[Link]([Link](true) ? 0 : 1);
}
}
Mapreduce Word Count
Mapper Code: You have to copy paste this program into the WCMapper Java
Class file.
// Importing libraries
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public class WCMapper extends MapReduceBase implements Mapper<LongWritable,
Text, Text, IntWritable> {
// Map function
public void map(LongWritable key, Text value, OutputCollector<Text,
IntWritable> output, Reporter rep) throws IOException
{
String line = [Link]();
// Splitting the line on spaces
for (String word : [Link](" "))
{
if ([Link]() > 0)
{
[Link](new Text(word), new IntWritable(1));
7
Adithya M (Dept. of CSE, CEC)
}
}
}
}
Reducer Code: You have to copy paste this program into the WCReducer
Java Class file.
// Importing libraries
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public class WCReducer extends MapReduceBase implements Reducer<Text,
IntWritable, Text, IntWritable> {
// Reduce function
public void reduce(Text key, Iterator<IntWritable> value,
OutputCollector<Text, IntWritable> output,
Reporter rep) throws IOException
{
int count = 0;
// Counting the frequency of each words
while ([Link]())
{
IntWritable i = [Link]();
count += [Link]();
}
[Link](key, new IntWritable(count));
}
}
Driver Code: You have to copy paste this program into the WCDriver Java
Class file.
8
Adithya M (Dept. of CSE, CEC)
// Importing libraries
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public class WCDriver extends Configured implements Tool {
public int run(String args[]) throws IOException
{
if ([Link] < 2)
{
[Link]("Please give valid inputs");
return -1;
}
JobConf conf = new JobConf([Link]);
[Link](conf, new Path(args[0]));
[Link](conf, new Path(args[1]));
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
[Link]([Link]);
[Link](conf);
return 0;
}
// Main Method
public static void main(String args[]) throws Exception
{
int exitCode = [Link](new WCDriver(), args);
9
Adithya M (Dept. of CSE, CEC)
[Link](exitCode);
}
}
[Link]
mapreduce-using-cloudera-distribution-hadoop-cdh/
[Link]
[Link]
[Link]
10
Adithya M (Dept. of CSE, CEC)