Find Top K IP Addresses that visited the website using MapReduce

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 :

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 17572

We 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 4353

I 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