Data Lab: December 2013

Thursday, 26 December 2013

In-mapper Combiner Program to Calculate Average

In my previous post I described how we can use in-mapper combiner to make our M/R program more efficient. In the post, we also saw both M/R algorithm for average calculation with/without using in-mapper combiner optimization.

In this post I am posting codes for both the algorithm.

The M/R program to calculate average without in-mapper combiner is given below:

	package com.hadoop.imcdp;

	import java.io.IOException;
	import java.util.HashMap;
	import java.util.Iterator;
	import java.util.Map;
	import java.util.Map.Entry;
	import java.util.Set;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.conf.Configured;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.io.FloatWritable;
	import org.apache.hadoop.io.IntWritable;
	import org.apache.hadoop.io.LongWritable;
	import org.apache.hadoop.io.Text;
	import org.apache.hadoop.mapreduce.Job;
	import org.apache.hadoop.mapreduce.Mapper;
	import org.apache.hadoop.mapreduce.Reducer;
	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;
	import org.apache.hadoop.util.Tool;
	import org.apache.hadoop.util.ToolRunner;

	public class AvgDriver extends Configured implements Tool{

	public static class ImcdpMap extends Mapper<LongWritable, Text, IntWritable, IntWritable> {

	String record;

	protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
	record = value.toString();
	String[] fields = record.split(",");

	Integer s_id = new Integer(fields[0]);
	Integer marks = new Integer(fields[2]);
	context.write(new IntWritable(s_id), new IntWritable(marks));
	} // end of map method
	} // end of mapper class


	public static class ImcdpReduce extends Reducer<IntWritable, IntWritable, IntWritable, FloatWritable> {

	protected void reduce(IntWritable key, Iterable<IntWritable> values, Reducer<IntWritable, IntWritable, IntWritable, FloatWritable>.Context context) throws IOException, InterruptedException {
	Integer s_id = key.get();
	Integer sum = 0;
	Integer cnt = 0;

	for (IntWritable value:values) {
	sum = sum + value.get();
	cnt = cnt + 1;
	}

	Float avg_m = (float) sum/cnt;
	context.write(new IntWritable(s_id), new FloatWritable(avg_m));
	}
	}

	@Override
	public int run(String[] args) throws Exception {
	Configuration conf = new Configuration();
	args = new GenericOptionsParser(conf, args).getRemainingArgs();
	String input = args[0];
	String output = args[1];

	Job job = new Job(conf, "Avg");
	job.setJarByClass(ImcdpMap.class);
	job.setInputFormatClass(TextInputFormat.class);
	job.setMapperClass(ImcdpMap.class);
	job.setMapOutputKeyClass(IntWritable.class);
	job.setMapOutputValueClass(IntWritable.class);

	job.setReducerClass(ImcdpReduce.class);
	job.setOutputFormatClass(TextOutputFormat.class);
	job.setOutputKeyClass(IntWritable.class);
	job.setOutputValueClass(FloatWritable.class);

	FileInputFormat.setInputPaths(job, new Path(input));
	Path outPath = new Path(output);
	FileOutputFormat.setOutputPath(job, outPath);
	outPath.getFileSystem(conf).delete(outPath, true);

	job.waitForCompletion(true);
	return (job.waitForCompletion(true) ? 0 : 1);
	}

	public static void main(String[] args) throws Exception {
	int exitCode = ToolRunner.run(new AvgDriver(), args);
	System.exit(exitCode);
	}
	}

view raw AvgDriver.java hosted with ❤ by GitHub

The M/R program to calculate average with in-mapper combiner is given below:

	package com.hadoop.imcdp;

	import java.io.IOException;
	import java.util.HashMap;
	import java.util.Iterator;
	import java.util.Map;
	import java.util.Map.Entry;
	import java.util.Set;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.conf.Configured;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.io.FloatWritable;
	import org.apache.hadoop.io.IntWritable;
	import org.apache.hadoop.io.LongWritable;
	import org.apache.hadoop.io.Text;
	import org.apache.hadoop.mapreduce.Job;
	import org.apache.hadoop.mapreduce.Mapper;
	import org.apache.hadoop.mapreduce.Reducer;
	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;
	import org.apache.hadoop.util.Tool;
	import org.apache.hadoop.util.ToolRunner;

	public class ImcdpAvgDriver extends Configured implements Tool{

	public static class ImcdpMap extends Mapper<LongWritable, Text, IntWritable, IntPair> {

	String record;
	Map partial_sum = new HashMap<Integer, Integer>();
	Map record_count = new HashMap<Integer, Integer>();

	protected void map(LongWritable key, Text value, Mapper.Context context) {
	record = value.toString();
	String[] fields = record.split(",");

	Integer s_id = new Integer(fields[0]);
	Integer marks = new Integer(fields[2]);

	if (partial_sum.containsKey(s_id)) {
	Integer sum = (Integer) partial_sum.get(s_id) + marks;
	partial_sum.put(s_id, sum);
	} else {
	partial_sum.put(s_id, marks);
	}

	if (record_count.containsKey(s_id)) {
	Integer count = (Integer) record_count.get(s_id) + 1;
	record_count.put(s_id, count);
	} else {
	record_count.put(s_id, 1);
	}
	} // end of map method

	protected void cleanup(Context context) throws IOException, InterruptedException {
	Iterator<Map.Entry<Integer, Integer>> itr1 = partial_sum.entrySet().iterator();

	while (itr1.hasNext()) {
	Entry<Integer, Integer> entry1 = itr1.next();
	Set record_count_set = record_count.entrySet();
	Integer s_id_1 = entry1.getKey();
	Integer partial_sum_1 = entry1.getValue();
	Integer record_count_1 = (Integer) record_count.get(s_id_1);

	context.write(new IntWritable(s_id_1), new IntPair(partial_sum_1, record_count_1));
	System.out.println(s_id_1+","+partial_sum_1+","+record_count_1);
	}
	} // end of cleanup
	} // end of mapper class


	public static class ImcdpReduce extends Reducer<IntWritable, IntPair, IntWritable, FloatWritable> {

	protected void reduce(IntWritable key, Iterable<IntPair> values, Reducer<IntWritable, IntPair, IntWritable, FloatWritable>.Context context) throws IOException, InterruptedException {
	Integer s_id = key.get();
	Integer sum = 0;
	Integer cnt = 0;
	System.out.println(key+","+values);
	for (IntPair value:values) {
	sum = sum + value.getFirstInt();
	cnt = cnt + value.getSecondInt();
	}

	System.out.println(sum+","+cnt);
	Float avg_m = (float) sum/cnt;
	context.write(new IntWritable(s_id), new FloatWritable(avg_m));
	}
	}

	@Override
	public int run(String[] args) throws Exception {
	Configuration conf = new Configuration();
	args = new GenericOptionsParser(conf, args).getRemainingArgs();
	String input = args[0];
	String output = args[1];

	Job job = new Job(conf, "IMCDP");
	job.setJarByClass(ImcdpMap.class);
	job.setInputFormatClass(TextInputFormat.class);
	job.setMapperClass(ImcdpMap.class);
	job.setMapOutputKeyClass(IntWritable.class);
	job.setMapOutputValueClass(IntPair.class);

	job.setReducerClass(ImcdpReduce.class);
	job.setOutputFormatClass(TextOutputFormat.class);
	job.setOutputKeyClass(IntWritable.class);
	job.setOutputValueClass(FloatWritable.class);

	FileInputFormat.setInputPaths(job, new Path(input));
	Path outPath = new Path(output);
	FileOutputFormat.setOutputPath(job, outPath);
	outPath.getFileSystem(conf).delete(outPath, true);

	job.waitForCompletion(true);
	return (job.waitForCompletion(true) ? 0 : 1);
	}

	public static void main(String[] args) throws Exception {
	int exitCode = ToolRunner.run(new ImcdpAvgDriver(), args);
	System.exit(exitCode);
	}
	}

view raw ImcdpAvgDriver.java hosted with ❤ by GitHub

The programs took 56sec and 42sec respectively for execution on my laptop for 10million records. So we can see a 33% improvement in time while using in-mapper combiner program.

In-Mapper Combiner

Recently I read a book on Map/Reduce algorithms by Lin and Dyer. This books gives a deep insight in designing efficient M/R algoriths. Today in this post I will post about in-mapper combining alogrithm and a sample M/R program using this algorithm.

Advantages of in-mapper combiner over traditional combiner:

When a mapper with a traditional combiner (the mini-reducer) emits the key-value pair, they are collected in the memory buffer and then the combiner aggregates a batch of these key-value pairs before sending them to the reducer. The drawbacks of this approach are:

The execution of combiner is not guaranteed; so MapReduce jobs cannot depend on the combiner execution.
Hadoop may store the key-value pairs in local filesystem, and run the combiner later which will cause expensive disk IO.
A combiner only combines data in the same buffer. Thus, we may still generate a lot of network traffic during the shuffle phase even if most of the keys from a single mapper are the same. To see this, consider the word count example, assuming that buffer size is 3, and we have <key, value> = <Stanford, 3>, <Berkeley, 1>, <Stanford, 7>, <Berkeley, 7>, and <Stanford, 2> emitted from one mapper. The first three items will be in one buffer, and last two will be in the the other buffer; as a result, the combiner will emit <Stanford, 10>, <Berkeley, 1>, <Berkeley, 7>, <Stanford, 2>. If we use in-mapper combiner, we will get <Stanford, 12>, <Berkeley, 8>.

Consider the case of calculating average marks for student. Let us consider we have below dataset

s_id    c_id    marks
8001    101    78
8001    102    88
8002    101    56
8002    102    77

The pseudo code for a basic M/R algorithm which computes average marks is as given:
class Mapper
    method Map(integer s_id, integer m)
        Emit(integer s_id, integer m)

class Reducer
    method Reduce(integer s_id, integer [m1 , m2 , . . .])
        sum ← 0
        cnt ← 0
        for all integer m ∈ integer [m1 , m2 , . . .] do
            sum ← sum + m
            cnt ← cnt + 1
        avg_m ← sum/cnt
        Emit(integer s_id, float avg_m )

If we have a large number of input records then the same number of records emitted from map task will be shuffled and sorted before being passed on to reducer. This large amount of data transfer could be deterrent in the speed of execution of overall M/R job.

We can make this algorithm faster by decreasing the number of records emitted by the mapper. To achieve this we can use an associative array to store partial sums of marks, and another associative array to store the count of marks and finally emit these values in close method. The pseduo code for in-mapper combiner is shown below:

class Mapper
    method Initialize
        S ← new AssociativeArray
        C ← new AssociativeArray
    method Map(integer s_id, integer m)
        S{s_id} ← S{s_id} + m
        C{s_id} ← C{s_id} + 1
    method Close
        for all integer s_id ∈ S do
            Emit(integer s_id, pair (S{s_id}, C{s_id}))

class Reducer
    method Reduce(integer s_id, pairs [(s1 , c1 ), (s2 , c2 ) . . .])
        sum ← 0
        cnt ← 0
    for all pair (s, c) ∈ pairs [(s1 , c1 ), (s2 , c2 ) . . .] do
            sum ← sum + s
            cnt ← cnt + c
    avg_m ← sum/cnt
    Emit(integer s_id, float avg_m )

Using this algorithm we can improve the performance of M/R job by reducing the number of intermediary key-value pairs emitted from mappers to reducers.

In my next post I shall post M/R program using in-mapper combiner and also compare the performance of this with M/R program without using any optimization.

Friday, 20 December 2013

Hive Data Types

Hive Data Types

Hive supports different data types to be used in table columns. The data types supported by Hive can be broadly classified in Primitive and Complex data types.

The primitive data types supported by Hive are listed below:

1. Numeric Types

TINYINT (1-byte signed integer, from -128 to 127)
SMALLINT (2-byte signed integer, from -32,768 to 32,767)
INT (4-byte signed integer, from -2,147,483,648 to 2,147,483,647)
BIGINT (8-byte signed integer, from -9,223,372,036,854,775,808 to 9,223,372,036,854,775,807)
FLOAT (4-byte single precision floating point number)
DOUBLE (8-byte double precision floating point number)
DECIMAL (Hive 0.13.0 introduced user definable precision and scale)

2. Date/Time Types

TIMESTAMP
DATE

3. String Types

STRING
VARCHAR
CHAR

4. Misc Types

BOOLEAN
BINARY

Apart from these primitive data types Hive offers some complex data types which are listed below:

5. Complex Types

arrays: ARRAY<data_type>
maps: MAP<primitive_type, data_type>
structs: STRUCT<col_name : data_type [COMMENT col_comment], ...>
union: UNIONTYPE<data_type, data_type, ...>

Wednesday, 18 December 2013

Logstash, ElasticSearch and Kibana Integration for Clickstream Weblog Ingestion

In this blog I am going to show case how we can develop a quick and easy demo application for clickstream weblog ingestion, search and visualization. We will achieve this using Logstash for log ingestion, store it in ElasticSearch and make a pretty dashboard using Kibana. For clickstream weblog I am using logs data from ECML/PKDD 2005 Discovery Challenge . You can download complete weblogs after registering there. These weblog are delimited by semi-colon (;) and have below mentioned fields in order:

shop_id
unixtime
client ip
session
visted page
referrer

Here are some sample log lines:

15;1075658406;212.96.166.162;052ecba084545d8348806f087b6e09bb;/ls/?&id=77&view=2,6,31&pozice=20;http://www.shop5.cz/ls/?id=77
12;1075658406;195.146.109.248;05aa4f4db0162e5723331042eb9ce8a7;/ct/?c=153;http://www.shop3.cz/
12;1075658407;212.65.194.144;86140090a2e102f1644f29e5ddadad9b;/ls/?id=34;http://www.shop3.cz/ct/?c=155
14;1075658407;80.188.85.210;f07f39ec63abf67f965684f3fa5729c4;/findp/?&id=63&view=1,2,3,14,20,15&p_14=nerez;http://www.shop4.cz/ls/?&p_14=nerez&id=63&view=1%2C2%2C3%2C14%2C20%2C15&&aktul=0
17;1075658408;194.108.232.234;be0970125c4eb3ee4fc380be05b3c58f;/ls/?id=155&sort=45;http://www.shop7.cz/ls/?id=155&sort=45
12;1075658409;62.24.70.41;851f20e644eb8bf82bfdbe4379050e2e;/txt/?c=734;http://www.shop3.cz/onakupu/

For creating this demo we need to create a logstash configuration file (lets name this file clickstream.conf) which consists of specifying inputs, filters and outputs. The clickstream.conf file looks like:

input { 
  file {
# path for clickstream log
    path => "/home/rishav.rohit/Desktop/clickstream/_2004_02_01_19_click_stream.log"
# define a type for all events handeled by this input
    type => "weblog"
    start_position => "beginning"
# the clickstream log is in character set ISO-8859-1
    codec => plain {charset => "ISO-8859-1"}
   }
}

filter {
  csv {
# define columns present in weblog
    columns => [shop_id, unixtime, client_ip, session, page, referrer]
    separator => ";"
  }
  grok {
# get visited page and page parameters
    match => ["page", "%{URIPATH:page_visited}(?:%{URIPARAM:page_params})?"]
     remove_field => ["page"]
  }
  date {
# as we are getting unixtime field in epoch seconds we will convert it to normal timestamp
    match => [ "unixtime", "UNIX" ]
  }
  geoip {
# this will convert ip to longitude-latitude using GeoLiteCity database from Maxmind
    source => "client_ip"
    fields => ["latitude","longitude"]
    target => "geoip"
    add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
    add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]
  }
  mutate {
# this will convert geoip.coordinates to float values
    convert => [ "[geoip][coordinates]", "float" ]
  }
}

output { 
# store output in local elasticsearch cluster
  elasticsearch {
    host => "127.0.0.1"
  }
}

To start logstash agent we run below command:

java -jar logstash-1.2.2-flatjar.jar agent -f clickstream.conf

A sample record in ElasticSearch looks like this:

{

    _index: logstash-2004.02.01
    _type: logs
    _id: I1N0MboUR0O1O3RZ-qXqnw
    _version: 1
    _score: 1
    _source: {
        message: [
            14;1075658407;80.188.85.210;f07f39ec63abf67f965684f3fa5729c4;/findp/?&id=63&view=1,2,3,14,20,15&p_14=nerez;http://www.shop4.cz/ls/?&p_14=nerez&id=63&view=1%2C2%2C3%2C14%2C20%2C15&&aktul=0 
        ]
        @timestamp: 2004-02-01T18:00:07.000Z
        @version: 1
        type: weblog
        host: HMECL000315.happiestminds.com
        path: /home/rishav.rohit/Desktop/clickstream/_2004_02_01_19_click_stream.log
        shop_id: 14
        unixtime: 1075658407
        client_ip: 80.188.85.210
        session: f07f39ec63abf67f965684f3fa5729c4
        referrer: http://www.shop4.cz/ls/?&p_14=nerez&id=63&view=1%2C2%2C3%2C14%2C20%2C15&&aktul=0
        page_visited: /findp/
        page_params: ?&id=63&view=1,2,3,14,20,15&p_14=nerez
        geoip: {
            latitude: 50.08330000000001
            longitude: 14.466700000000003
            coordinates: [
                14.466700000000003
                50.08330000000001
            ]
        }
    }

}

So we have parsed complex log message into simpler components and converted fields like unixtime to datetime, IP to latitude-longitude and got page visited by the client. Now using Kibana we can quickly make dashboard with these panels

This histogram shows page landings count for different time interval

This is a map pointing to client locations

And in this table we can see different attributes for each clickstream