Friday, January 13, 2012

Clustering of sparse data using python with scikit-learn

Coming from a Matlab background, I found sparse matrices to be easy to use and well integrated into the language. However, when transitioning to python’s scientific computing ecosystem, I had a harder time using sparse matrices. This post is intended to help Matlab refugees and those interested in using sparse matricies in python (in particular, for clustering)

Requirements:
scikit-learn (2.10+)
numpy (refer to scikit-learn version requirements)
scipy (refer to scikit-learn version requirements)

Sparse Matrix Types:
There are six types of sparse matrices implemented under scipy:
  1. bsr_matrix -- block sparse row matrix
  2. coo_matrix -- sparse matrix in coordinate format
  3. csc_matrix -- compressed sparse column matrix
  4. csr_matrix -- compressed sparse row matrix
  5. dia_matrix -- sparse matrix with diagonal storage
  6. dok_matrix -- dictionary of keys based sparse matrix
  7. lil_matrix -- row-based linked list sparse matrix
For more info see: (http://docs.scipy.org/doc/scipy/reference/sparse.html):

When to use which matrix:
The following are scenarios when you would want to choose one sparse matrix type over the another:
  • Fast Arithmetic Operation: csc_matrix, csr_matrix
  • Fast Column Slicing (e.g., A[:, 1:2]): csc_matrix
  • Fast Row Slicing (e.g., A[1:2, :]) csr_matrix
  • Fast Matrix vector products: csr_matrix, bsr_matrix, csc_matrix
  • Fast Changing of sparsity (e.g., adding entries to matrix): lil_matrix, dok_matrix
  • Fast conversion to other sparse formats: coo_matrix
  • Constructing Large Sparse Matrices: coo_matrix
Clustering with scikit-learn:
With the release of scikit-learn 2.10, one of the useful new features is the support for sparse matrices with the k-means algorithm. The following is how you would use sparse matrices with k-means:

Full Matrix to Sparse Matrix

Example-1
-----------------------------------------------------------------------------------------------
from numpy.random import random
from scipy.sparse import *
from sklearn.cluster import KMeans

# create a 30x1000 dense matrix random matrix.
D = random((30,1000))
# keep entries with value < 0.10 (10% of entries in matrix will be non-zero)
# X is a "full" matrix that is intrinsically sparse.
X = D*(D<0.10) # note: element wise mult

# convert D into a sparse matrix (type coo_matrix)
# note: we can initialize any type of sparse matrix.
# There is no particular motivation behind using
# coo_matrix for this example.
S = coo_matrix(X)

labeler = KMeans(k=3)
# convert coo to csr format
# note: Kmeans currently only works with CSR type sparse matrix
labeler.fit(S.tocsr())

# print cluster assignments for each row
for (row, label) in enumerate(labeler.labels_):
print "row %d has label %d"%(row, label)
-----------------------------------------------------------------------------------------------

One of the issues with Example-1 is that we are constructing a sparse matrix from a full matrix. It will often be the case that we will not be able to fit a full (although intrinsically sparse) matrix in memory. For example, if the matrix X was a 100000x1000000000 full matrix, there could be some issues. One solution to this is to somehow extract out the non-zero entries of X and to use a smarter constructor for the sparse matrix.

Sparse Matrix Construction

In Example-2, we will assume that we have X's data stored on some file on disk. In particular, we will assume that X is stored in a csv file and that we are able to extract out the non-zero data efficiently.

Example-2
-------------------------------------------------
import csv
from scipy.sparse import *
from sklearn.cluster import KMeans

def extract_nonzero(fname):
"""
extracts nonzero entries from a csv file
input: fname (str) -- path to csv file
output: generator<(int, int, float)> -- generator
producing 3-tuple containing (row-index, column-index, data)
"""
for (rindex,row) in enumerate(csv.reader(open(fname))):
for (cindex, data) in enumerate(row):
if data!="0":
yield (rindex, cindex, float(data))

def get_dimensions(fname):
"""
determines the dimension of a csv file
input: fname (str) -- path to csv file
output: (nrows, ncols) -- tuple containing row x col data
"""

rowgen = (row for row in csv.reader(open(fname)))
# compute col size
colsize = len(rowgen.next())
# compute row size
rowsize = 1 + sum(1 for row in rowgen)
return (rowsize, colsize)

# obtain dimensions of data
(rdim, cdim) = get_dimensions("X.csv")

# allocate a lil_matrix of size (rdim by cdim)
# note: lil_matrix is used since we be modifying
# the matrix a lot.

S = lil_matrix((rdim, cdim))

# add data to S
for (i,j,d) in extract_nonzero("X.csv"):
S[i,j] = d

# perform clustering
labeler = KMeans(k=3)
# convert lil to csr format
# note: Kmeans currently only works with CSR type sparse matrix
labeler.fit(S.tocsr())

# print cluster assignments for each row
for (row, label) in enumerate(labeler.labels_):
print "row %d has label %d"%(row, label)
--------------------------------------------------


What to do when Sparse Matrices aren't supported:
When sparse matrices aren't supported, one solution is to convert the matrix to a full matrix. To do this, simply invoke the todense() method.

Hudson/Jenkins With RVM and PhantomJS

Setting up Hudson/Jenkins to work RVM (Ruby Version Manager) and PhantomJS (for headless JavaScript testing) can be painful. This post will show you how to easily set them up on your own server.

RVM
At Bizo we have several projects that have dependencies on different versions of Ruby, mostly due to some projects relying on older gems which are incompatible with Ruby 1.9. Installing RVM on a dev machine is almost always a cinch but getting it to play nicely with your ci build server isn't quite so straightforward.


We run our Hudson server off of an Amazon EC2 instance. Our EC2 instances are started up with custom software, but  it really boils down to executing a bash start up script. Assuming the Hudson user's $HOME is set to /var/lib/hudson, you can copy/paste  the code below to install RVM for you. Otherwise just replace /var/lib/hudson below to the $HOME dir of your Hudson (or Jenkins) user.


# RVM
COMMANDS=$(cat <<EOS
bash -s stable < <(curl -s https://raw.github.com/wayneeseguin/rvm/master/binscripts/rvm-installer)
echo "[ -s \"/var/lib/hudson/.rvm/scripts/rvm\" ] && source \"/var/lib/hudson/.rvm/scripts/rvm\" # loads RVM" > .bashrc
 
# ensure RVM is loaded
source ~/.bashrc

echo "Installing Ruby 1.9.2"
rvm install 1.9.2 
rvm use 1.9.2

echo "Installing gems for Ruby: 1.9.2"
gem install bundler --no-rdoc --no-ri
 
# add additional ruby versions here
EOS
)
su - hudson -c "$COMMANDS"

Then in your Hudson build go to your project configuration and under "execute shell" you can invoke rvm and run your project like normal. Note* our version of Hudson doesn't automatically load .bashrc, so you might need to source it first to ensure RVM loads, ex:

source ~/.bashrc
 
# Pick our ruby version
rvm use 1.9.2

# Run your project... ex bundle install && rake test:units for a Rails project 


PhantomJS
PhantomJS is our execution environment of choice for running JavaScript unit tests and setting it up to run on Hudson is actually quite easy.

Here is the necessary bash snippet to make it available for use in Hudson. 


INSTALL_PATH= # wherever you want
wget http://phantomjs.googlecode.com/files/phantomjs-1.4.1-linux-x86-dynamic.tar.gz ${INSTALL_PATH}/phantomjs.tar.gz
# OR For 64bit machines wget http://phantomjs.googlecode.com/files/phantomjs-1.4.1-linux-x86_64-dynamic.tar.gz

mkdir ${INSTALL_PATH}/phantomjs
tar -zxvf ${INSTALL_PATH}/phantomjs.tar.gz -C ${INSTALL_PATH}

ln -s ${INSTALL_PATH}/phantomjs/bin/phantomjs /usr/local/bin/phantomjs
 
After running the script above you can invoke PhantomJS as "phantomjs" in the "execute shell" box inside your project configuration. You'll probably want your tests to fail with a non-0 exit status so the Hudson build will fail, if you use the Jasmine testing framework you can use our phantom-jasmine test runner on Github: https://github.com/jcarver989/phantom-jasmine

Interactive Hive sessions, Elastic MapReduce, and GNU screen

One extremely annoying quality of using Hive interactively on EMR (or any other remote system) is that your sessions will die if you lose your connection to the server.  Once this happens, your ssh session will end, terminating both your Hive session and any queries that may currently be running.

In most cases, this happens when I'm waiting for a query to execute and I need to move from one place to another, whether from my desk to a conference room or from the office to home.  When I can predict (or know) that I'm going to lose my connection and just want to be able to reconnect to Hive later, the best option I've found is to run Hive inside of GNU screen.

I'm definitely a screen newbie, but there are really only three things you need to know:

1. As soon as you log in for the first time, install screen and start it up:

sudo apt-get install screen
screen
hive

2. When you're (temporarily) done interacting with Hive and want to stick it in the background, tell screen to detach Hive from your current session by pressing "Ctrl-a" then "d".  You may now log out from your EMR node.

3. When you're ready to resume your Hive session, simply log back on and tell screen to reconnect to the most recent session:

screen -r

Screen does a whole lot of other stuff, but simply allowing graceful reconnection to Hive sessions is definitely worth the price of entry.

For comparison, some other things you could do to work around this problem are using nohup, suspending/putting the job in the background and using disown, or using an even more advanced tool like  tmux.

Monday, January 9, 2012

How to Measure Latency Distribution using Amazon CloudWatch

By default, web services hosted in AWS behind an Elastic Load Balancer have their response rates automatically tracked in CloudWatch.  This lets you easily monitor the minimum, maximum, and average latency of your services.

Unfortunately, none of these metrics are very useful.  Minimum and maximum latencies really measure outliers, and averages can easily obscure what's really going on in your services.

To get a better picture of the performance of our backend services, we explicitly have client services track the latency distribution of servers using CloudWatch custom metrics.  Instead of having a single metric that measures the number of milliseconds taken by each request, we instead count the number of requests that take a particular number of milliseconds.

More specifically, for each service, we create eleven custom metrics representing the buckets 0-10ms, 11-20ms, 21-30ms, ... , 91-100ms, and 101+ms.  (We actually have another set of buckets covering the 100-1000ms range at 100ms intervals, but we've found these to be less useful.)  For each request, we then increment the counter for the bucket corresponding to the request's latency.  Periodically, an automated process simply pulls the aggregated counts from CloudWatch into a Google Spreadsheet and graph the results in a stacked bar chart.

A stacked bar chart of response times for one of our services.  Each layer represents a different response time bucket.

This gives us a nice visual representation of both the overall traffic levels as well as how many of them had response times below each threshold.

Our automated process also converts the totals into percentiles.

A stacked bar chart of response times as percentiles.

Google's chart tools support displaying only a vertical slice of the data, so we can easily show 90th, 95th, and 97.5th percentiles of our response times.  This lets us easily see whether we're fulfilling performance requirements like "95% of all requests with response times below N ms."

Our slowest responses.

The combination of latency bucketing, Amazon CloudWatch, and Google Spreadsheets gives us a very lightweight way of tracking our server performance.  The only additional overhead on our servers is a bit of logic to do local aggregation and push data into CloudWatch, and the only other moving part is a simple cron task that connects together the CloudWatch and Google Data APIs.

Monday, December 19, 2011

Get Your Company To Blog More With A Game

At Bizo we try to blog fairly often. But writing blog posts with any degree of frequency at a startup is tough - there are often ten other important tasks that needed to be done yesterday. Finding the time to sit down and write a post when you have features to build, code to review and the occasional meeting is difficult to say the least.

We needed something that would encourage us to blog more frequently, and there's no better way to motivate a bunch of engineers than a game. So during our last Open Source Day we built a Blog Scoreboard that ranks authors based on the number of posts and comments they have. It's setup on our office big screen tv serving as constant reminder that Larry is by a landslide, the king of blogging. For a live demo of the scoreboard and to see just how much Larry is dominating us, checkout the live example here.

Currently we just optimize towards the volume of posts so the points are currently assigned as: 10 points per blog post and 1 point per comment. The weighting scheme is intentionally naive due since this had to be built in a few hours and will undoubtably change as time goes on

The best part about our scoreboard is that it is an open source sinatra (ruby) app, and it works with any Blogger blog! All you have to do is edit a few lines of YAML and you'll have your very own big screen blog scoreboard. You can grab the source code and install instructions on github: https://github.com/jcarver989/blog-scoreboard Happy blogging!

Thursday, December 15, 2011

4 tips from the trenches of Amazon Elastic MapReduce and Hive

Some things you can learn from others, and some things you can only learn from experience.  In an effort to move some knowledge from the latter category into the former, here are four things we've learned the hard way while working with big data flows using Hive on Amazon's Elastic MapReduce:

1. To prevent data loss, make sure Hive knows if it owns the data.

All of our data processing follows a simple pattern: servers write logs to S3, and we use the basic EMR/S3 integration to read this data in our Hive scripts.  A typical table definition could look something like this:

create external table sample_data(d map<string,string>)
  comment 'data logs uploaded from servers'
  row format delimited
    fields terminated by '\004'
    collection items terminated by '\001'
    map keys terminated by '\002'
  stored as textfile
  location 's3://log-bucket/hive/data/files/sample_data/'
;

(The single map column is our way of supporting dynamic columns in Hive.)

The most easily overlooked part of this table definition is the "external" keyword, which tells Hive that the actual underlying data is managed by some other process.  If you forget to add this keyword and later issue a "drop table" command, Hive will happily nuke all of your log files.

This can be especially troublesome while doing ad hoc analysis, as these usually involve interactive queries in the Hive console with a workflow that often involves an exploratory style of development that includes deleting and recreating tables.

One common use pattern where dropping tables appears is when running scripts that include a create statement.  Normally, trying to recreate a table that already exists in Hive will cause an error, so a script may preemptively issue a drop command in case that table already exists.  An alternative is to change the table definition to tell Hive to ignore create statements for preexisting tables:

create external table if not exists sample_data(d map<string,string>)

In the event that you accidentally forget to specify "external" in your table definition, you can add it later by altering the table:

alter table sample_data set tblproperties ('EXTERNAL'='TRUE') ;

Note that the capitalization in the table properties is significant.  Additionally, this feature is only available in Hive 0.6.0+.  (It was present but buggy prior to 0.6.0.)

2. Use a smart partitioning scheme to create flexible views of the data.

Hive supports partitioning, which can be a huge performance win when querying only a subset of the entire dataset.  This is critical when the entire dataset consists of years of request information but you're only interested in analyzing one day of traffic.  To take advantage of this, we upload our data into hourly paths in S3:

s3://log-bucket/log-name/year=${year}/month=${month}/day=${day}/hour=${hour}/${logfile}.log.gz

We would then load this data with the following statements:

create external table sample_data(d map<string,string>)
  comment 'data logs uploaded from servers'
  partitioned by (
    year string, 
    month string, 
    day string, 
    hour string)
  row format delimited
    fields terminated by '\004'
    collection items terminated by '\001'
    map keys terminated by '\002'
  stored as textfile
  location 's3://log-bucket/log-name/'
;
alter table sample_data recover partitions ;

The year/month/day/hour partitions are now available to be used in select/where statements just like columns.

We could (and originally did) just use a single partition column for the entire date; however, using multiple partition columns allows Hive to complete ignore the presence of other partitions.  The above statement will need to recover metadata for 8-9 thousand partitions of data, which (while less troublesome than not partitioning at all) will still require a lot of time and memory.  Multiple partition columns lets us create (for example) a "view" of a single month:

create external table sample_data_2011_12(d map<string,string>)
  comment 'data logs uploaded from servers'
  partitioned by (
    day string, 
    hour string)
  row format delimited
    fields terminated by '\004'
    collection items terminated by '\001'
    map keys terminated by '\002'
  stored as textfile
  location 's3://log-bucket/log-name/year=2011/month=12/'
;
alter table sample_data recover partitions ;

Now Hive only needs to recover 7-8 hundred partitions.  We use a similar strategy for our daily reporting, which makes recovering partition data even faster.

The only time this scheme breaks down is when the report boundaries don't align with the partition boundaries.  In these cases, you can still get the benefits of partitioning by manually adding the partition information to the table.  For example, to do efficient queries on a week of data, we would replace the "recover partitions" statement with a sequence of statements like these (and tweak the table definition to use only a single partition column):

alter table sample_data add partition(day_hour=2011120600) location "s3://log-bucket/log-name/year=2011/month=12/day=06/hour=00/";

3. Use S3 multipart uploads for large objects.

While S3, Elastic MapReduce, and Hive theoretically allow you easily scale your data storage and analytics, we regularly run up against operational limits as our processing requirements grow.  One surprising problem we recently ran into was S3 throttling EMR because our cluster was accessing the source data too quickly.  After some back-and-forth with support, they suggested a workaround of uploading the source objects (which were multiple GB in size and created by a previous EMR job) with multi-part uploads.

Enabling multi-part uploads in EMR is simply a matter of flipping a configuration switch in the cluster configuration.  When starting a cluster from the command line, simply add the following options (taken from the linked documentation):

elastic-mapreduce --create --alive \
--bootstrap-action s3://elasticmapreduce/bootstrap-actions/configure-hadoop \
--bootstrap-name "enable multipart upload" \
--args "-c,fs.s3n.multipart.uploads.enabled=true, \
-c,fs.s3n.multipart.uploads.split.size=524288000"

4. Spot instances can dramatically reduce your data costs.

As our data processing requirements have grown, so have our cluster sizes.  We've found that we can reduce our costs by over 50% by using a mix of on-demand and spot instances.

Each Elastic MapReduce job can have up to three instance groups: master, core, and task.  All data resides on the master and core instances, so jobs can use spot instances to add processing power and additional IO with very little risk.  You can add these instances at startup by simply specifying a bid price:

elastic-mapreduce --create --alive --hive-interactive --hive-versions 0.7 \
--instance-group master --instance-type m1.xlarge --instance-count 1 \
--instance-group core --instance-type m1.xlarge --instance-count 9 \
--instance-group task --instance-type m1.xlarge --instance-count 90 --bid-price 0.25 

For ad hoc queries, I've also found it easy to prototype my job on a subset of data using only the master and core instances, then add the task instances once I'm satisfied that the query is correct.  New task instances are available as Hadoop workers as soon as they're added to the cluster, even if you're already running your MapReduce job.  You can modify the size of the task group using the command-line client:

elastic-mapreduce --jobflow j-xxxxxxxxxxxxx --add-instance-group task  --instance-type m1.xlarge --instance-count 90 --bid-price 0.25

elastic-mapreduce --jobflow j-xxxxxxxxxxxxx --set-num-task-group-instances 90

Note that while the size of the task group can be changed, you cannot modify the instance type or the bid price after the task group is created.

Tuesday, December 13, 2011

Promises in Javascript/Coffeescript

This happens often. You’re humming along writing some awesome Javascript code. At first everything is neat and organized. Then you add a feature here, an AJAX call there and before your once lovely codebase has turned into callback spaghetti.

One scenario where this can become particularly nasty is when you have an arbitrary number of asynchronous tasks, one of which cant run until all the others have completed. Suppose for example you are creating a page that displays information on your organization’s engineering team and open source projects hosted on Github. This might require a few api calls to Github (eg. one to get the repositories, another to get the team members) that can be executed in parallel. We could just append the information to the DOM as each api call finishes, but it would be nice to avoid pieces of the site popping in at different times. Instead it would be ideal if we made the api calls to Github and only after both api calls had completed would we render the information to the DOM. The problem is that in Javascript trying to implement this can get quite messy.

A Bad Solution

The issue here is knowing when it is safe to execute the task that renders the Github information to the page. One strategy might be to store the results from each api call in an array and then wait for a set amount of time before drawing the page.

results = []

githubApiCallOne (response) -> 
  results.push(response)

githubApiCallTwo (response) ->
  results.push(response)

setTimeout(() ->
  drawPage(results)
, 5000)

This turns out to be a poor solution as we can either end up waiting too little, in which case the program would fail or we could set the timeout too high making our users wait longer than necessary to load our page.

Promises to The Rescue

Fortunately there is a better way to implement our Github page. Using a construct called a Promise (sometimes called a Future) allows us to elegantly handle these types of situations. Using promises we can turn our code into something like this:

Promise.when(
  githubApiCallOne(),
  githubApiCallTwo()
).then((apiCallOneData, apiCallTwoData) ->
  renderPage(apiCallOneData, apiCallTwoData)
)

The basic idea is that our async api calls will now return a promise object that functions much like an IOU – they can’t give us the results of the api call immediately but they (probably) can at some time in the future. The Promise.when method takes an arbitrary number of promise objects as parameters and then executes the callback in the “then” method once every promise passed to “when” has been completed.

To do this, our api calls would have to be modified to return promise objects, which turns out to be trivial. Such an implementation might look like so:

githubApiCallOne = () ->
  promise = new Promise()

  # async call
  ajaxGet("/repositories", (repository_data) ->
    # fulfill the promise when async call completes
    promise.complete(repository_data)
  )

  return promise

githubApiCallTwo = () ->
  promise = new Promise()

  ajaxGet("/users", (user_data) ->
    promise.complete(user_data)
  )

  return promise

The githubApiCallOne and githubApiCallTwo make their ajax calls but return a promise object immediately. Then when the AJAX calls complete, they can fulfill the promise objects by calling “complete” and passing in their data. Once both promise objects have been fulfilled the callback passed to Promise.then is executed and we render the page. With jQuery

The good news is if you’re already using jQuery you get Promises for free. As of jQuery 1.5 all the $.ajax methods (eg. $.get, $.post etc) return promises which allows you to do this:

promise1 = $.get "http://foo.com"
promise2 = $.post "http://boo.com"

$.when(promise1, promise2)
 .then (promise1Result, promise2Result) ->
  # do something with the data

What if I cant use jQuery?

Rolling a custom implementation of Promises isn’t recommended for production code but might be necessary if you write a lot of 3rd party Javascript and/or just want to try it for fun. Here’s a very basic implementation to get you started. Error handling, exceptions etc are left as an exercise to the reader.

class Promise
  @when: (tasks...) ->
    num_uncompleted = tasks.length 
    args = new Array(num_uncompleted)
    promise = new Promise()

    for task, task_id in tasks
      ((task_id) ->
        task.then(() ->
          args[task_id] = Array.prototype.slice.call(arguments)
          num_uncompleted--
          promise.complete.apply(promise, args) if num_uncompleted == 0
        )
      )(task_id)

    return promise
    
  constructor: () ->
    @completed = false
    @callbacks = []

  complete: () ->
    @completed = true
    @data = arguments
    for callback in @callbacks
      callback.apply callback, arguments

  then: (callback) ->
    if @completed == true
      callback.apply callback, @data
      return

    @callbacks.push callback

Sharp eyed readers might notice that the code inside the for loop in the Promise.when method looks a bit strange. You might notice that I’m wrapping the promise’s “then” method call inside of a self executing function that passes in the task_id variable. This funkiness is actually required due to the way that closures work in Javascript. If you attempt to reference the task_id without the self executing closure, you’ll actually get a reference to the task_id iterator instead of a copy – which means by the time your “then” methods execute the loop will have finished iterating and all the task_ids will share the same value! To get around this you have to create a new scope and pass in the iterator so we end up with a copy of the value instead of a reference.

And Finally an example using the supplied Promise class to prove it works:

delay = (string) ->
  promise = new Promise()
  setTimeout(() -> 
    promise.complete string
  ,200)
  return promise

logEverything = (fooData, barData, bazData) -> 
  console.log fooData[0], barData[0], bazData[0]

window.onload = () ->
  Promise.when(
    delay("foo"),
    delay("bar"),
    delay("baz")
  ).then logEverything