Seek patterns of Elasticsearch

One must know the disk seek patterns of an application when optimizing the storage layer for any application. So when I was working on the performance analysis of ES, one of the first thing was to determine its disk seek patterns. I used blktrace and seekwatcher for that purpose.

blktrace is a utility that traces block io request issued to a given block device. To quote from the blktrace manual:

…blktrace receives data from the kernel in buffers passed up through the debug file system (relay). Each device being traced has a file created in the mounted directory for the debugfs, which defaults to /sys/kernel/debug

For using blktrace, the first step is to mount debugfs. It is a memory based file system to debug linux kernel code.

# mount -t debugfs debugfs /sys/kernel/debug

The next step is to setup the disk on which we need to the tracing. This should be a separate disk as we do not want other OS activity to influence our readings. For this, I’ve used /dev/sdc as the data directory of ES. Now, start ES and wait for a few seconds after the cluster comes into green state, so as to be sure that we are only tracing the disk activity while searching and not the startup. Before firing the ES query, start blktrace on /dev/sdc with the following command

blktrace -d /dev/sdc -o es-ssd-search

This will start tracing the block requests on the disk and send output to es-ssd-search.blktrace.0 .. es-ssd-search.blktrace.n-1 where each file represents requests from one core. Since I was using a quadcore CPU, I got the following files:

es-on-ssd.blktrace.0
es-on-ssd.blktrace.1
es-on-ssd.blktrace.2
es-on-ssd.blktrace.3

Now that we have data from blktrace, the next step is to visualize it. That can be done by blkparse utility. It formats the blktrace output into human readable form. But as they say, a picture is worth a thousand words, there is another tool, namely seekwatcher that can produce plots and movies from the output of blktrace. I used it to visualize the seek patterns. To encode movies with seekwatcher, we also need to install MEncoder. Once seekwatcher and MEncoder are installed, run the following command to generate the movie:

# seekwatcher --movie-frames=10 -m -t es-on-ssd.blktrace.0 -o es-on-ssd.movie

It will produce es-on-ssd.movie that can be played with MPlayer. Following is the output that I got:

As can be seen, apart from a few random reads most of the reads are sequential-reads for which I went on to optimize the storage.

Reference:

Bonus video: ES startup seeks

ES Start Seek Patterns from Anand Nalya on Vimeo.

Clustering of synthetic control data in R

This is an R implementation for clustering example provided with Mahuot. The orignal problem description is:

A time series of control charts needs to be clustered into their close knit groups. The data set we use is synthetic and so resembles real world information in an anonymized format. It contains six different classes (Normal, Cyclic, Increasing trend, Decreasing trend, Upward shift, Downward shift). With these trends occurring on the input data set, the Mahout clustering algorithm will cluster the data into their corresponding class buckets. At the end of this example, you’ll get to learn how to perform clustering using Mahout.

We will be doing the same but using R instead of Mahout. The input dataset is available here.

For running this example, in addition to R, you also need to install the flexclust package available from CRAN. It provides a number of methods for clustering and cluster-visualization.

Here is the script:

x <- read.table("synthetic_control.data")
cat( "read", length(x[,1]), "records.\n")

# load clustering library
library(flexclust)

# get number of clusters from user
n <- as.integer( readline("Enter number of clusters: ")) 

# run kmeans clustering on the dataset
cl1 <- cclust(x, n)

print("clustering complete")

# show summary of clustering
summary(cl1)

# plot the clusters
plot(cl1, main="Clusters")

readline("Press enter for cluster histogram")
m<-info(cl1, "size") # size of each cluster
hist(rep(1:n, m), breaks=0:n, xlab="Cluster No.", main="Cluster Plot")

readline("Press enter for a plot of distance of data points from its cluster centorid")
stripes(cl1)
print("done")

Here are the graphs produced when we run the above script with no. of clusters, n=7

Clusters

clusters

Frequency Histogram

frequency

Distance from centroid

centroid distance

Building a Single Page Webapp with jQuery

In a typical single page web app, you load a html page and keep updating it by requesting data from the server either periodically or whenever the user asks for. Such an application is well suited for data driven systems. For example, if you already have a REST API, you can easily create a single page web client for it. The best way of understanding what a single page web app is by looking at the web interface of Gmail

So let us define what we want to achieve in the end with our application:

  • We should be able to load different views on the same page.
  • We should be able to translate user intentions (clicks) into appropriate javascript function calls.
  • The url shown in the address bar of the browser should represent the state of the application. That is, when the user refreshes the browser, it should have the same content as prior to refresh.

The Root Page

So let us begin by defining the basic structure of our application. index.html is our application page that will be modified during the lifetime of our application. For the sake of simplicity, let us assume that we will be making changes to #application div only.

Intercepting link clicks

Now we need to intercept all the click actions on the links and change the url hash accordingly. For handling the url hashes, we are using the Jquery BBQ plugin. Intercepting the link clicks will help us in preventing the default navigation to other pages. Also, as this is the required behaviour for all the links, we will setup the interceptor at global level. If required you can change the scope of this interception by modifying the jQuery selector where we define the interception.

Defining the Routes

Next, we need to define the routes that are mapped to url hash. Also, since each hashchange represents an action on part of user, we will also define what function to execute for each hashchange. We do this by listening for haschange event and firing the appropriate js function.

Initialization

Even though, we have proper routes and actions defined now, our index page is still empty. We now need to initialize the app. This is done by setting the default hash, which will fire the hashchange event when the page loads or in case the user has refreshed the page, just triggering the hashchange event manually.

Sample Action

Now that we have the skeleton for our application ready, lets have a look at a sample action. Suppose the default action, is showing a list of users [#/users]. We have mapped this hash to the function showUserList. Now there are several ways of combining the html structure and the json data for userList, but we will be using the simplest of methods. First we will get the html fragment representing the userList using the $.load function, and then we will populate it with actual data that we get from the REST api.

var showUserList = function(){
	$('#app').load('/html/users.html #userListDiv', function() {
		$.getJSON('/users', function(data) {
			
			// create the user list
			var items = [ '<ul>'];
			$.each(data, function(index, item) {
				items.push('<li><a href="/users/"' + item.id + '">'
						+ item.name + '</a></li>');
			});
			items.push('</ul>');
			
			var result = items.join('');
			
			// clear current user list
			$('#userListDiv').html('');
			
			// add new user list
			$(result).appendTo('#userListDiv');
		});
	}
};

If you find all the string concatenation going on in above snippet a bit sloppy, you can always use your favourite template plugin for jQuery.

Note: The Object.extended function in above snippets comes from the wonderful Sugar.js library that makes working with native javascript objects a breeze.

Idea for Windows Phone Team – Make switching easier and less costlier

Microsoft is already paying developers for porting their applications from iOS and Android. This is all well for the developers but what about the end users. They have already invested a lot in the apps on those platform. So the cost of switching is not just the hardware/carrier-contract cost but also buying all those apps again for WP7. Also, what happens to all the data that is stored in the apps on other platforms?

Microsoft, here is an idea for: pay extra to developer to enable importing data from other platforms. And make all those apps free for the consumers who already purchased those apps on some other platform.

Role based views using css and javascript

In web applications, role based views are normally done using server side if..else tags etc. When developing pure javascript clients, one way of doing role based views is again using javascript conditionals. This approach suffers from the fact that role-related code is scattered throughout the document. Another simple way to achieve the same is using css classes.

Let us assume that we have three roles, ADMIN and USER and GUEST. Then we will define three css classes as follows:

Now consider the following html doc, that has css classes attached to elements according to the current user’s role.

At this stage if you load the document in a browser, you will not see anything as nothing all the roles are invisible. So lets spruce this up with a bit of javascript to set proper css class properties.

If a cookie role is set that represents the current user’s role then adding the above snippet of code will enable all the elements tagged with css class .role_{roleCookie} to be visible.

Even though this is a very simple implementation, we can modify it easily to take into account more complex scenarios. For example, if you have completely ordered role based visibility (access level), i.e. given role R1 and R2, we can always tell which role has more visibility, then we can extend as follows:

$(document).ready(function() {
    // We are assuming that jQuery and jQuery Cookie plugin are included in the doc

    var role = $.cookie('user_role'); // Should return one of 'ADMIN', 'USER' OR 'GUEST'

    // set property for relevant css class
    if(role == "ADMIN") {
        // make everything visible
        $("<style type='text/css'>.role_admin, .role_user, .role_guest {display:block} </style>").appendTo("head");

    } else if(app.role == "USER") {
        // make user and guest part visible, admin part will remain invisible
        $("<style type='text/css'>.role_user, .role_guest {display:block} </style>").appendTo("head");

    } else if(app.role == "GUEST") {
        // only show the guest part
        $("<style type='text/css'>.role_guest {display:block} </style>").appendTo("head");
    }
});

Caution: One thing to note here is that all the parts of the document are sent to all the users, only what is visible on the browser is role based. So, server side authorisation checks will always be there as nothing stops a suspecting user from looking into the source and firing that dreaded request.

Hadoop Tuning Notes

This is a quick dump of my notes for hadoop tuning.

General

  • Hadoop is designed to use multiple cores and disks, so it will be able to take full advantage of more powerful hardware.
  • Don’t use RAID for HDFS datanodes as redundancy is handled by HDFS. RAID should be used for namenodes as it provides protection against metadata corruption.
  • Machine running the namenode should run on a 64-bit system to avoid 3GB limit on JVM heap size.
  • In case the cluster consists of more than one rack, it is recommended to tell Hadoop about the network topology. Rack awareness will help Hadoop in calculating data locality while assigning MapReduce tasks and it will also help HDFS to choose replica location for files more intelligently.
  • Two processors will be engaged by datanode and tasktracker, and the remaining n-2 processors can have a factor of 1 to 2 extra jobs.
  • On Master node, each of namenode, jobtracker and secondary namenode takes 1000M of memory. If you have a large number of files than increase the JVM heap size for namenode and secondary namenode.

Configuration Parameters

HDFS

dfs.block.size
The block size used by HDFS which defaults to 64MB. On large clusters this can be increased to 128MB or 256MB to reduce memory requirements on namenode and also to increase the size of data given to map tasks.
dfs.name.dir
should give a list of directories where the namenode persists copies of data. It should be one or two local disks and a remote disk such as nfs mounted directory so that in case of node failure, metadata can be recovered from the remote disk.
dfs.data.dir
specifies the list of directories used by datanodes to store data. It should always be local disks and if there are multiple disks then each directory should be on different disk so as to maximize parallel read and writes.
fs.checkpoint.dir
list of directories where secondary namenode keeps checkpoints. It should use redundant disks for the same reason as dfs.name.dir
dfs.replication
number of copies of data to be maintained. It should be at least 2 more than the number of machines that are expected to fail everyday in the cluster.
dfs.access.time.precision
The precision in msec that access times are maintained. It this value is 0, no access time are maintained resulting in performance boosts. Also, Storage disks should be mounted with noatime which disables last access time updates during file reads resulting in considerable performance gains.
dfs.datanode.handler.count
Number of threads handling block requests. In case on multiple physical disks, the throughput can increase by increasing this number from default value of 3.
dfs.namenode.handler.count
Number of threads on Namenode. This number should be increase from default value of 10 for large clusters.

MapReduce

mapred.local.dir
is the list of directories where intermediate data and working files are store by the tasks. These should be a number of local disks to facilitate parallel IO. Also, these partitions should the same which are used by datanodes to store data (dfs.data.dir)
mapred.tasktracker.map.tasks.maximum, mapred.tasktracker.reduce.tasks.maximum
specifies the maximum number of map and reduce tasks that can be run at the same time. It should be a multiple of number of cores.
mapred.job.tracker.handler.count
Number of server threads for handling tasktracker request. Default value is 10 and the recommended value is 4% of the tasktracker nodes.
mapred.child.java.opts
increase the JVM heap memory of tasks that require more memory.

Others

core-site.xml::io.file.buffer.size
This is buffer size used by hadoop during IO which default to 4KB. On modern system it can be increased to 64KB or 128KB for performance gains.

Building a simple [Yahoo] S4 application

S4 is a distributed stream processing platform from Yahoo. It is often seen as the real-time counterpart of Hadoop. S4 being fault tolerant and horizontally scalable helps you in building very large stream processing application that can do anything from detecting earthquakes to finding that perfect bit of advertising that the visitor on your website is most likely to click.

At its core, an S4 application consists of a number of Processing Elements (PEs) that are wired together with the help of a spring configuration file that defines the PEs and the flow of events in the system. Also, events are produced by event producers that listen that sends these events to the client adapter for S4, from where, the S4 platform takes over and dispatch it to appropriate processing elements. After processing these events, PEs can choose to dispatch them to other PEs for further processing or they can choose to produce output events. Thus, arbitrarily complex behavior can be derived together by wiring a simple set of PEs.

S4 comes with a few example applications, but here is a much simpler S4WordCount application that shows how to:

  1. Keep state in a PE.
  2. Dispatch events from a PE.
  3. Process multiple events from a single PE.
  4. Write a simple java client for sending events to S4.

S4 is a distributed stream processing platform from Yahoo. It is often seen as the real-time counterpart of Hadoop. S4 being fault tolerant and horizontally scalable helps you in building very large stream processing application that can do anything from detecting earthquakes to finding that perfect bit of advertising that the visitor on your website is most likely to click.

At its core, an S4 application consists of a number of Processing Elements (PEs) that are wired together with the help of a spring configuration file that defines the PEs and the flow of events in the system. Also, events are produced by event producers that listen that sends these events to the client adapter for S4, from where, the S4 platform takes over and dispatch it to appropriate processing elements. After processing these events, PEs can choose to dispatch them to other PEs for further processing or they can choose to produce output events. Thus, arbitrarily complex behavior can be derived together by wiring a simple set of PEs.

S4 comes with a few example applications, but here is a much simpler S4WordCount application that shows how to:

  1. Keep state in a PE.
  2. Dispatch events from a PE.
  3. Process multiple events from a single PE.
  4. Write a simple java client for sending events to S4.

In S4WordCount, we will build a simple WordReceiverPE, that will receive events in the form of word and will simply print these words on stdout. It will also identify sentences in the word stream and then forward these sentences for further processing to SenteceReceiverPE. WordReceiverPE will also produce receive sentence events and print them to stdout.

First lets have a look at Word and Sentence, the event object used in our example:

package test.s4;

public class Word {
	private String string;

	public String getString() {
		return string;
	}

	public void setString(String message) {
		this.string = message;
	}

	@Override
	public String toString() {
		return "Word [string=" + string + "]";
	}
}

S4 uses keys, which is a set of some properties of the event object, for routing/dispatching events. In this case since Word is the key-less entry point into the system, and we don’t have any key for it, but for Sentence which will be processed further, we have a Sentence.sentenceId as the key. (For simplicity, all Sentence have the same sentenceId, 1)

Now let’s have a look at our first PE, i.e. WordReceiverPE:

We define a StringBuilder that will be used to accumulate words to form a Sentence. The processEvent(Word) method simply prints the received word on stdout and appends the word to the builder. It then checks if the sentence is complete by checking for . at the end of the builder and if so, it creates a Sentence event(object) and dispatches it to the Sentence stream. Once dispatched the processEvent(Sentence) will receive that event and will again print the sentence to stdout.

Now let’s have a look at SenteceReceiverPE, our second PE which does nothing but print the received Sentence on stdout.

Finally, lets see the content of the application config file that wires all this together and forms a valid S4 application. The name of the config file should follow the naming convention of <AppName>-config.xml. In this case we will call the config file S4WordCount-conf.xml

This is a spring bean definition file. The first bean, wordCatcher is an object of class test.s4.WordReceiverPE and it injects a dispatcher called dispatcher into WordRecieverPE. We will look at properties of dispatcher later. keys define the stream on which our PE will be listening for event. RawWords * means that it will be receiving all events on the stream named RawWords irrespective of their keys. Similar is the intention for Sentence *.

Our second bean is sentenceCatcher which is an object of class test.s4.SentenceReceiverPE and will accept all events on stream called Sentence.

Third is the definition for the dispatcher which we injected into wordCatcher. The dispatcher needs a partitioner that partitions the events based on some key and then dispatches them to appropriate PEs. In this case we are using the DefaultPartitioner whose properties are defined later by sentenceIdPartitoner bean which says that partition the event objects on Sentence stream by sentenceId property. dispatcher uses the S4 provided commLayerEmitter to emit the events.

Running the application

To run this application on the S4:

  1. Setup S4 as documented here.
  2. Create a S4WordCount.jar from above classes.
  3. Deploy the application on S4, by creating the following directory structure:
    	/$S4_IMAGE
    		/s4-apps
    			/S4WordCount
    				S4WordCount-conf.xml
    				/lib
    					S4Wordcount.jar
  4. Start S4: $S4_IMAGE/scripts/start-s4.sh -r client-adapter &
  5. Start client adapter: $S4_IMAGE/scripts/run-client-adapter.sh -s client-adapter -g s4 -d $S4_IMAGE/s4-core/conf/default/client-stub-conf.xml &

Now S4 is ready to receive events. Following is an event sender that uses the java client library to send events to S4. It reads one word at a time from stdin and sends it to S4.

Run client: java TestMessageSender localhost 2334 RawWords test.s4.Word

Here is the sample output on the S4 console:


Received: Word [string=this]
Received: Word [string=is]
Received: Word [string=a]
Received: Word [string=sentence.]

Using fast path!
Value 1, partition id 0
wrapper is stream:Sentence keys:[{sentenceId = 1}:0] keyNames:null event:Sentence [string= this is a sentence.] => 0
Received Sentence(WordReceiverPE) : Sentence [string= this is a sentence.]
Received Sentence: Sentence [string= this is a sentence.]

Curious case of Lewis Carrol, tournament runner ups and sorting

No less than four grand slams are played every year to find the best tennis player in the world. These are knockout tournaments in which, at each stage half of the players lose and get out. Two [supposedly] best players clash in the finals and the winner of that game is declared the champion (the best player) and the loser the runner up (the second best player). All is fine with our champion but there is just a small problem with our runner up: He might not be the second best player of the tournament.

How?

Let us assume that there were 16 players in the beginning with seedings 1 to 16, 16 being the best. Consider the following results for the matches in the tournament

15
	15
1
		15
5
	9
9
			16
12
	16
16
		16
4
	4
3
				16
11
	11
6
		11
7
	10
10
			14
2
	8
8
		14
13
	14
14

As expected, 16 is the winner but the runner up is 14 not 15. 15 crashed out in the semi finals playing against the champion.

So what can we conclude from this?

  1. Runner up may not be the second best player of the tournament.
  2. log n matches are required to decide the best player in the tournament.
  3. We can find the second best player in the tournament looking at the match results of the winner which is a list of log n players. At some point he must have defeated the second best player.

Here is a program that gives you the second best player in (n - 1) + ((log n) - 1) = n + (log n) - 2 comparisons: