Today, I will demonstrate how to find top K items from given data based on count. The data set that I am using looks like this :
Here we will write 2 MR jobs to do following:
Job1:
Find (IP Addresses , Total count) that shows how many times that IP Address visited the website .
Job2:
Find Top 10 (IP Address , TotalCount) based on highest count in descending order.
Code Job1:
Mapper Code -> We are reading line by line and writing context(IPAddress,1)
Reducer Code -> Iterate values and aggregate all values for each IP Address .
Code Job2:
Job2 will take input data from job1 output. This is defined in driver code in following line :
FileInputFormat.setInputPaths(job2, out);
In the mapper part of Job2, we have created a TreeMap object that will hold count and IPAddress .
Let's try to run this code by uncommenting following line in driver code for job2:
job2.setNumReduceTasks(0);
This will make sure , no reducer code is running for Job2.
Job2's Mapper is splitting the output of Job1 on tab separator. We are putting the count as key and Ipaddress as value in TreeMap.In the cleanup code, we are making sure that count column becomes the key and IPAddress becomes the value. The reason for this operation is that it will sort based on column key (Count column since we want it in descending order) and send it to reducer.
One trick that I have done in Mapper is following 3 lines of code.As soon as count of records in treeMap is greater than 10, it will remove the item with smallest count from treeMap. This will make sure that at any given time, we will have Top 10 highest count keys in TreeMap.
Reducer is only changing the key/value pairs while writing it to output. The output we need should have a IPAddress as key and Count as value.
199.72.81.55 - - [01/Jul/1995:00:00:01 -0400] "GET /history/apollo/ HTTP/1.0" 200 6245 unicomp6.unicomp.net - - [01/Jul/1995:00:00:06 -0400] "GET /shuttle/countdown/ HTTP/1.0" 200 3985 199.120.110.21 - - [01/Jul/1995:00:00:09 -0400] "GET /shuttle/missions/sts-73/mission-sts-73.html HTTP/1.0" 200 4085 burger.letters.com - - [01/Jul/1995:00:00:11 -0400] "GET /shuttle/countdown/liftoff.html HTTP/1.0" 304 0 199.120.110.21 - - [01/Jul/1995:00:00:11 -0400] "GET /shuttle/missions/sts-73/sts-73-patch-small.gif HTTP/1.0" 200 4179 burger.letters.com - - [01/Jul/1995:00:00:12 -0400] "GET /images/NASA-logosmall.gif HTTP/1.0" 304 0 burger.letters.com - - [01/Jul/1995:00:00:12 -0400] "GET /shuttle/countdown/video/livevideo.gif HTTP/1.0" 200 0 205.212.115.106 - - [01/Jul/1995:00:00:12 -0400] "GET /shuttle/countdown/countdown.html HTTP/1.0" 200 3985
Here we will write 2 MR jobs to do following:
Job1:
Find (IP Addresses , Total count) that shows how many times that IP Address visited the website .
Job2:
Find Top 10 (IP Address , TotalCount) based on highest count in descending order.
Code Job1:
Mapper Code -> We are reading line by line and writing context(IPAddress,1)
Reducer Code -> Iterate values and aggregate all values for each IP Address .
public static class WLMapper1 extends
Mapper<LongWritable, Text, Text, IntWritable> {
// HashMap<String, Integer> map = new HashMap<String, Integer>();
@Override
protected void map(LongWritable key, Text value, Context context)
throws java.io.IOException, InterruptedException {
String IPorServerName = "";
if (value.toString().contains(" ")) {
String[] arr = value.toString().split(" ");
if (arr.length > 0) {
IPorServerName = arr[0].toLowerCase().trim();
}
}
context.write(new Text(IPorServerName), new IntWritable(1));
};
} public static class WLReducer1 extends
Reducer<Text, IntWritable, Text, IntWritable> {
@Override
protected void reduce(Text key, Iterable<IntWritable> values,
Context context) throws java.io.IOException,
InterruptedException {
int count = 0;
if (values.iterator().hasNext()) {
for (IntWritable x : values) {
int val = new Integer(x.toString());
count += 1;
} context.write(key, new IntWritable(count));
}
};
}
When you run this code , following output will be generated for Job1:
007.thegap.com 45 01-dynamic-c.rotterdam.luna.net 1 01-dynamic-c.wokingham.luna.net 28 02-dynamic-c.wokingham.luna.net 13 03-dynamic-c.wokingham.luna.net 15 04-dynamic-c.rotterdam.luna.net 22 04-dynamic-c.wokingham.luna.net 28 05-dynamic-c.rotterdam.luna.net 26 05-dynamic-c.wokingham.luna.net 9 06-dynamic-c.rotterdam.luna.net 22 07-dynamic-c.rotterdam.luna.net 4 0772jela.jelke.rpslmc.edu 3 08-dynamic-c.rotterdam.luna.net 4
Code Job2:
Job2 will take input data from job1 output. This is defined in driver code in following line :
FileInputFormat.setInputPaths(job2, out);
In the mapper part of Job2, we have created a TreeMap object that will hold count and IPAddress .
Let's try to run this code by uncommenting following line in driver code for job2:
job2.setNumReduceTasks(0);
This will make sure , no reducer code is running for Job2.
public static class WLMapper2 extends
Mapper<LongWritable, Text, IntWritable, Text> {
// Our output key and value Writables
private TreeMap<Integer, String> repToRecordMap = new TreeMap<Integer, String>();
@Override
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// Parse the input string into a nice map
if (value.toString().contains("\t")) {
String[] arr = value.toString().split("\t");
if (arr.length > 1) {
repToRecordMap.put(Integer.parseInt(arr[1]), arr[0]);
if (repToRecordMap.size() > 10) {
repToRecordMap.remove(repToRecordMap.firstKey());
}
}
}
}
@Override
protected void cleanup(Context context) throws IOException,
InterruptedException {
for (Map.Entry<Integer, String> entry : repToRecordMap.entrySet()) {
Integer key = entry.getKey();
String value = entry.getValue();
context.write(new IntWritable(key), new Text(value));
}
}
}
Here is the output if we make number of reducers=0 for Job2.
4353 disarray.demon.co.uk 4863 news.ti.com 4906 163.206.89.4 5434 edams.ksc.nasa.gov 5922 piweba2y.prodigy.com 7573 siltb10.orl.mmc.com 7852 alyssa.prodigy.com 9868 piweba1y.prodigy.com 11591 piweba4y.prodigy.com 17572 piweba3y.prodigy.com
Job2's Mapper is splitting the output of Job1 on tab separator. We are putting the count as key and Ipaddress as value in TreeMap.In the cleanup code, we are making sure that count column becomes the key and IPAddress becomes the value. The reason for this operation is that it will sort based on column key (Count column since we want it in descending order) and send it to reducer.
One trick that I have done in Mapper is following 3 lines of code.As soon as count of records in treeMap is greater than 10, it will remove the item with smallest count from treeMap. This will make sure that at any given time, we will have Top 10 highest count keys in TreeMap.
if (repToRecordMap.size() > 10) {
repToRecordMap.remove(repToRecordMap.firstKey());
}
Reducer is only changing the key/value pairs while writing it to output. The output we need should have a IPAddress as key and Count as value.
public static class WLReducer2 extends
Reducer<IntWritable, Text, Text, IntWritable> {
@Override
protected void reduce(IntWritable key, Iterable<Text> values,
Context context) throws IOException, InterruptedException {
for (Text x : values) {
context.write(new Text(x.toString()), key);
}
};
}
Lets run the code and see if we can get the output as expected :
disarray.demon.co.uk 4353 news.ti.com 4863 163.206.89.4 4906 edams.ksc.nasa.gov 5434 piweba2y.prodigy.com 5922 siltb10.orl.mmc.com 7573 alyssa.prodigy.com 7852 piweba1y.prodigy.com 9868 piweba4y.prodigy.com 11591 piweba3y.prodigy.com 17572We are getting the correct output but the problem is that we wanted to show the output sorted in descending order on count column. To achieve this, we need to write a comparator class:
public static class KeyComparator extends WritableComparator {
protected KeyComparator() {
super(IntWritable.class, true);
}
@Override
public int compare(WritableComparable w1, WritableComparable w2) {
// TODO Auto-generated method stub
// Logger.error("--------------------------> writing Keycompare data = ----------->");
IntWritable ip1 = (IntWritable) w1;
IntWritable ip2 = (IntWritable) w2;
int cmp = -1 * ip1.compareTo(ip2);
return cmp;
}
The above class can be used by adding following line in driver class.job2.setSortComparatorClass(KeyComparator.class);
Now the output looks like :
piweba3y.prodigy.com 17572 piweba4y.prodigy.com 11591 piweba1y.prodigy.com 9868 alyssa.prodigy.com 7852 siltb10.orl.mmc.com 7573 piweba2y.prodigy.com 5922 edams.ksc.nasa.gov 5434 163.206.89.4 4906 news.ti.com 4863 disarray.demon.co.uk 4353I am using following code in driver to make sure that job1 is successful before I start job2:
boolean succ = job.waitForCompletion(true);
if (!succ) {
System.out.println("Job1 failed, exiting");
return -1;
}
here is the code for driver class that I have used to chain these 2 jobs.
public int run(String[] args) throws Exception {
// TODO Auto-generated method stub
Configuration conf = getConf();
Job job = new Job(conf, "WL Demo");
job.setJarByClass(WLDemo.class);
job.setMapperClass(WLMapper1.class);
job.setReducerClass(WLReducer1.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
Path in = new Path(args[0]);
Path out = new Path(args[1]);
Path out2 = new Path(args[2]);
FileInputFormat.setInputPaths(job, in);
FileOutputFormat.setOutputPath(job, out);
boolean succ = job.waitForCompletion(true);
if (!succ) {
System.out.println("Job1 failed, exiting");
return -1;
}
Job job2 = new Job(conf, "top-k-pass-2");
FileInputFormat.setInputPaths(job2, out);
FileOutputFormat.setOutputPath(job2, out2);
job2.setJarByClass(WLDemo.class);
job2.setMapperClass(WLMapper2.class);
job2.setReducerClass(WLReducer2.class);
// job2.setReducerClass(Reducer1.class);
job2.setInputFormatClass(TextInputFormat.class);
job2.setMapOutputKeyClass(IntWritable.class);
job2.setMapOutputValueClass(Text.class);
job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(IntWritable.class);
job2.setSortComparatorClass(KeyComparator.class);
// job2.setNumReduceTasks(0);
succ = job2.waitForCompletion(true);
if (!succ) {
System.out.println("Job2 failed, exiting");
return -1;
}
return 0;
}
Bug Found and fixed:One of the issues identified with this approach is that when 2 items have same count, TreeMap will ignore one of the items which will provide incorrect output. To solve this , I have changed the treemap to following :
1) In Mapper 2 , use :
private TreeMap<Integer, List<String>> records = new TreeMap<Integer, List<String>>();
Use following code to restrict total count to less than desired count :
records.put(Integer.parseInt(arr[1]), keyVal);
if (records.size() > Top_k_Items) {
records.remove(records.firstKey());
}
But still total count can be greater than desired count as treemap have a list as value. This will be taken care in Reducer 2 like this :
public static class WLReducer2 extends
Reducer<IntWritable, Text, Text, IntWritable> {
int count = 0;
@Override
protected void reduce(IntWritable key, Iterable<Text> values,
Context context) throws IOException, InterruptedException {
for (Text x : values) {
if (count < Top_k_Items) {
context.write(new Text(x), key);
count = count + 1;
}
}
};
Complete code for solving this problem can be found at this link.
Please provide your feedback in case if you find any issues with this code.
Happy Hadooping !!!
Have a great day.
Varun
