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
 

Read Multiple CSV files using lapply, rbind & reduce functions - R programming

let's do a quick demo on how lapply works in R programming. Suppose you have a number of csv files and each csv file have same structure.
Here is sample data in 1 csv .

"Date","sulfate","nitrate","ID"
"2009-01-01",NA,NA,69
"2009-01-02",NA,NA,69
"2009-01-03",NA,NA,69
"2009-01-04",NA,NA,69
"2009-01-05",NA,NA,69
"2009-01-06",NA,NA,69

Now you want to read all csv files and calculate mean for 1 column. In this case , let's calculate mean  of "sulphate" column across number of different files.

Step 1: read multiple files and create dataframes.

files<- list.file()



 // here files is collection of all files

//  read.csv is the function that you want to apply on all files

dataFrames <- lapply(files, read.csv)

Step 2: combine all dataframes into 1 single dataframe

dataFrames <- Reduce(function(x, y) rbind(x, y), dataFrames)

Step 3: Calculate mean on dataframe column

mean(dataFrame[,sulphate], na.rm = TRUE)

 

Hive UDF implementation

Consider you have a data set like following:

AccidentId, Date, Day

 

1, 1/1/1979, 5               

 

2, 1/2/1979, 6

 

3, 1/3/1979, 7

 

4, 1/4/1979, 1

 

5, 1/5/1979, 2

 

6, 1/6/1979, 3

 

7, 1/7/1979, 4

 

8, 1/8/1979, 5

 

......

 

.......

 

16, 1/6/1980, 3

 

17, 1/7/1980, 4

 

18, 1/8/1980, 5

This is the query that we want to run on above written input data.

 Select Date , Day_of_Week,count(*) from accidents group by GetYear_FromDate(Date) , Day_of_Week;

The problem is that from Date column we just want Year part.

We have 2 ways to solve this problem :

1) By converting string date to proper format and then using Year function in Hive.

Year(cast(concat(substr(Date,7,4), '-',substr(Date,1,2), '-',substr(Date,4,2))as date)) as DT

2) Write  UDF to get following output:

1979    1    843
1979    2    1180
1979    3    1131
1979    4    1190
1979    5    1231
1979    6    1458
1979    7    1167
1989    7    2

We need to follow 2 rules for HIVE UDFs:
 1) A UDF must be subclass of  org.apache.hadoop.hive.ql.exec.UDF;
 2) A UDF must have 1 evaluate() method.

Here is the UDF code:

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

public class DateFix extends UDF {

    public Text evaluate(Text str) {
        if (str == null) {
            return null;
        }
       
        return new Text(str.toString().substring(6, 10));
    }
} 

 Now we need to register the UDF :

ADD JAR /home/edureka/Desktop/Projects/HiveUDf/DateDemo.jar;      

We also need to create alias for Java classname:

create temporary function GetYear_FromDate as 'DateFix';    

Finally, we run following query :

Select GetYear_FromDate(Date) , Day_of_Week,count(*) from accidents group by GetYear_FromDate(Date) , Day_of_Week;

That's it. You have just created a UDF in Hive and you can use it on any date column to fetch year part.

Thanks

How to implement WritableComparable Interface ?

Today, I decided to write a demo on how to implement customwritable ?

Let's say we have a dataset like following:

AccidentId, Date, Day

1, 1/1/1979, 5                

2, 1/2/1979, 6

3, 1/3/1979, 7

4, 1/4/1979, 1

5, 1/5/1979, 2

6, 1/6/1979, 3

7, 1/7/1979, 4

8, 1/8/1979, 5

......

.......

16, 1/6/1980, 3

17, 1/7/1980, 4

18, 1/8/1980, 5

 Here Day 5 means Thursday, 6 means Friday, 1 means Sunday and so on.

We want to calculate number of accidents on each day of week in each year. The output looks like :

Number of accidents per day in the year 1979

As you can see the reducer is emitting key which is the combination of year and day in the sample data. This is where we need 'WritableComparable' interface.

public class DOW implements WritableComparable<DOW> {

 private Text year;
 private Text day;

 // private int count;
 public DOW() {
  this.year = new Text();
  this.day = new Text();
  // this.count = count;
 }

 public DOW(Text year, Text day) {
  this.year = year;
  this.day = day;
  // this.count = count;
 }

 public Text getYear() {
  return this.year;
 }

 public void setYear(Text year) {
  this.year = year;
 }

 public Text getDay() {
  return this.day;
 }

 public void setDay(Text day) {
  this.day = day;
 }

 @Override
 public void readFields(DataInput in) throws IOException {
  // TODO Auto-generated method stub
  year.readFields(in);
  day.readFields(in);

 }

 @Override
 public void write(DataOutput out) throws IOException {
  // TODO Auto-generated method stub
  year.write(out);
  day.write(out);
 }

 @Override
 public int compareTo(DOW o) {
  // TODO Auto-generated method stub
  int cmp = year.compareTo(o.year);
  if (cmp != 0) {
   return cmp;
  }
  return day.compareTo(o.day);
 }

 @Override
 public String toString() {
  // TODO Auto-generated method stub
  return year + "," + day;
 }

 @Override
 public int hashCode() {
  // TODO Auto-generated method stub
  return year.hashCode() * 163 + day.hashCode();
 }
}

Once we have implemented the WritableComparable interface , we will use the class in Mapper and reducer code as follows.

In the mapper code, I am splitting the value on ',' and getting the year part out of date column. Also, I need day column from dataset .  Once I have both keys, I am writing the context as :

    context.write(new DOW(new Text(Integer.toString(year)), new Text(Integer.toString(day))),
       new IntWritable(1));

Key ->    new DOW(new Text(Integer.toString(year)), new Text(Integer.toString(day))
Value -> new IntWritable(1)

protected void map(LongWritable key, Text value, Context context)
    throws java.io.IOException, InterruptedException {

   if (value.toString().contains(",")) {
    String[] array = value.toString().split(",");
    if (!array[9].equals("Date")) {
     Date dt = null;
     try {
      dt = new SimpleDateFormat("dd/mm/yyyy").parse(array[9]);

     } catch (ParseException e) {
      // TODO Auto-generated catch block

      e.printStackTrace();
     }

     int year = dt.getYear();

     int day = Integer.parseInt(array[10].toString());
     sLogger.info("year=" + year);
     sLogger.info("array[10]=" + array[10]);
     context.write(new DOW(new Text(Integer.toString(year)),
       new Text(Integer.toString(day))),
       new IntWritable(1));
    }
   }

Reducer is simply calculating the number of accidents based on key.

@Override
  protected void reduce(DOW key, Iterable<IntWritable> values,
    Context context) throws java.io.IOException,
    InterruptedException {
   int count = 0;
   sLogger.info("key =" + key);
   for (IntWritable x : values) {
    int val = Integer.parseInt(x.toString());
    count = count + val;
   }
   context.write(key, new IntWritable(count));
  };

Once you run this code, you will be able to see the output based on the key year and day.

And that is it .

Happy Hadooping !!

Varun

User Defined Counters in Hadoop

Friends,

Today, we will learn few concepts about User Defined Counters. Hadoop have many counters such as :
1) MR Task Counters
2) File System Counters
3) File input & File Output format counters
4) Job Counters
5) User Defined Counters

Let's try to understand why do we need counters ?
Consider a scenario where you have many bad records in your input file. You want to find out how many such bad records exists . Ex: you are writing MR job for the query:
Select name,count(*) from User but some of the records are bad as the name do not exists in the columns . You would want to filter out such records and at the same time want to see how many such records exists. Here is the sample data :

Id,Name
1,Tom
2,John
3,
4,null
5,
6,Tom
7,Abi
8,Jack
9,Jack
10,

Let's write some code to create the counters. We will start by writing an enum to define the categories of records.

public enum RecordType {
  Good, Bad
 }

Here is the mapper that will increment the counter when it gets Good/Bad records.

protected void map(LongWritable key, Text value, Context context)
    throws java.io.IOException, InterruptedException {
   if (value.toString() != "" && value.toString().contains(",")) {

    String[] array = value.toString().split(",");

    if (array.length < 2) {
     context.getCounter(RecordType.Bad).increment(1);
    } else {
     if (array[1].isEmpty()) {

      context.getCounter(RecordType.Bad).increment(1);
     } else {
      context.getCounter(RecordType.Good).increment(1);
      context.write(new Text(array[1]), new IntWritable(1));
     }
    }

   }

  };

The reducer code is simple  and needs no explanation.

protected void reduce(Text key, Iterable<IntWritable> values,
    Context context) throws java.io.IOException,
    InterruptedException {
   int count = 0;

   for (IntWritable x : values) {
    int val = Integer.parseInt(x.toString());

    count += val;

   }
   context.write(key, new IntWritable(count));
  };

Once you run the code on the sample dataset . You should be able to see the following counters:

This looks good but the only issue is the name (CDemo$RecordType) is not too user friendly.
Lets see how we can change the names to better user friendly names.

1) Create a file with the name ClassName_Enum.properties . ex: CDemo_RecordType.properties
2) place the file at same location where your class exists
3) Enter following lines :

CounterGroupName= Counter records
Good.name= GOOD Records
Bad.name=BAD Records

4) Save the file and create the jar and run it using hadoop command.
5) Now you can see counters with the names supplied in your file.

Here is the source code :

https://github.com/varunkhatri/HadoopCounters/blob/master/CountersDemo/src/CDemo.java

Hope you enjoyed the demo.

Happy Hadooping !!!

Varun