#map-reduce

Thinking Web Scale-1: Map-Reduce – Bring compute to data

In the last decades and a half, there has arisen a class of problem that are becoming very critical in the computing domain. These problems deal with computing in a highly distributed environments. A key characteristic of this domain is the need to grow elastically with increasing workloads while tolerating failures without missing a beat.  In short I would like to refer to this as ‘Web Scale…

What the above graph shows that how a map-reduce is running in Hadoop.

Some details about that:

(1) An application starts a master instance and worker instances for the Map phase and, later, worker instances for the Reduce phase. 

(2) The master partitions the input data in segments.

(3) Each Map instance reads its input data segment and processes the data. During process, if any of Map instance fails, Master will assign its work to other workers from latest periodic  checkpoint.

(4) The results of the processing are stored on the local disks of the servers where the Map instances run.  After the Map workers finish, the location of data will be returned to master so that master can partition these data to Reduce workers.

(5) When all Map instances have finished processing their data, the Reduce instances read the results of the first phase and merge the partial results.

(6) The final results are written by the Reduce instances to a shared storage server.

I take no embarrassment in admitting that the term "Dynamic Programming" troubled me for many years. I couldn't figure how these mysterious algorithms worked. It's not that I didn't get the idea of divide and conquer or that of caching. But it seemed too easy that something as that is called "Dynamic Programming" would simply be a combination of "Divide and Conquer" and "Caching prior results of sub-problems". Then I came across this SO post: Dynamic Programming. 

This reaffirmed my suspicion that Dynamic Programming isn't a mystical creature that lives deep in the woods, but instead it is a combination of two techniques. Something that any astute algorithm developer would intuitively come to when writing such an algorithm. Divide and Conquer and then cache the prior results (if that be worth it). Mystery solved. 

Enter Map-Reduce.

If you had a computational problem of scale, and access to multiple processors, what would you do? Wouldn't you divide the problem into sub-parts and branch them off to the processors and recombine the result? 

Isn't this, in a simplistic sense, what Map-Reduce does? Sure, there are all the frameworks built upon this idea that facilitate this, but those are mostly implementation details.

At inagist.com we have been using Riak and yes we are loving it. We moved away from Cassandra after it started taxing our limited resources. The nice thing about Cassandra was the data model. Super columns allowed us to store metadata for a resource as needed. For example the retweets and replies of a tweet were stored in their own super columns associated with a tweet and we could pull it out as needed. Concurrency issues were also not a bother. We could do simultaneous updates to columns and super columns and not worry about data consistency issues. This is seriously tricky when maintaining tweet statistics. Popular tweets keep getting retweeted and replied to concurrently by many people. 

When looking for alternatives Riak was our first choice primarily because of it being in Erlang and since it had a map-reduce option which looked seriously promising. The ability to have a choice of backends was another compelling factor. Here are some of the interesting stuff that we have worked out in using Riak.

Using the Data model to our advantage 

At the heart of Riak everything is a key-value. All metadata is associated with the value and has to be read and updated as a single unit. The most interesting metadata is of-course the Links that you store along with a keys value. Interesting because Riak's map reduce has an extra option called link walking. This allows you to filter the links on a document by tag or bucket and feed the linked documents to the next phase. Infact Riak's map-reduce allows you to have any combination of link, map, reduce options to process your data. And yes these are optional too. So infact you can have a link-reduce, link-link-reduce or link-link type queries too. 

Why is this interesting? It allows us to store metadata on a seperate resource and link it to the main resource. Meaning we could have say 10 buckets storing the ids of the retweets as Links and the main resource has a link to these 10 buckets. You could parse through this list of with a link-link query. This reduces the contention on one resource for updates, parallelizes the read and spreads it across the cluster. We store replies and retweet details of a tweet in this model.

A link has three attributes Bucket, Key and Tag. Its supposed to refer to a Bucket and Key if you intend to further get data out of the linked document. But if you know what you are upto this allows to do for some serious extra data management. We currently store some tweet meta-data in Bucket and Key with a well known tag. When we later want to query for all replies to a

 tweet we do a link-link-reduce on well known tags and get the replies or retweets out. I'm not getting into specifics but it should give you the idea.

Interfacing with Riak

Most references point to using Riak via the HTTP interface or via the protocol buffers client. Great if you are working from a non Erlang environment.

We currently use the built in client for Riak over the protocol buffers client. With the main processing being in Erlang, and being a distributed app at that, there was no point in going thru extra layers to get into Riak. This also gives us some interesting options, like for example the "Your Friends" tab on the conversations page that you see once you log into inagist.com. This does a link-link-reduce-reduce where an extra reduce talks to remote erlang process for the logged in user to filter out only replies from his followers. See it in action on a popular tweet like this one from @BarackObama. Mind you the "Your Friends" feature will work only after your account is enabled on inagist.com, but the Popular tab works for anyone and pulls out only the replies which are of interest.

Storage options

And yes the back-end, currently we run the innostore back-end based on Embedded Innodb. This kind of makes Riak a distribution layer over a trusted storage layer. Of the back-ends available this has worked best for us giving a consistent performance along with being reasonable on the resource usage. But the biggest factor here is that it gives an option to plugin what you want like the trial we did with Tokyo Cabinet.

Our biggest bottle neck now is the disk space, we keep pruning the data set at a failry fast pace, roughly one week of data is all that we hold. We get a little above 5 million tweets a day from the twitter pipe and we keep cleaning out as the disks fill up.

My last post felt a little incomplete without some code backing it up. I'm following it up with a code sample of how exactly this map reduce is wired up. 

I will walk through how we do the "Popular Replies" section on the conversation page. Again here is a @BarackObama tweet, with more than a 500 replies. Popular replies extracts only those replies which have been further replied to, re-tweeted or a reply from the author of the tweet itself. Right now its picked out 1 of these 500+ replies.

Data Model

Resonses to a tweet are captured in a bucket of its own <<"tweet_responses_bucket">>. Each tweet is keyed by its tweet id as a 128 bit binary <<TweetId:128>>. Response details are not stored directly on this resource but a linked value in a bucket called "tweet_responses_subkeys_bucket". Responses are stored as links on a resource keyed as <<TweetId:128, (ResponseId rem 10):8>> in this bucket. This resource is added as a link on the {<<"tweet_responses_bucket">>, <<TweetId:128>>} resource and tagged as <<"tweet_response">>. A reply is recorded as a link of the form {{<<ResponseId:128>>, <<ResponseAuthorId:128>>}, <<"reply">>}. A link is represented as {{Bucket, Key}, Tag}, this link does not point to a valid bucket, key pair but is purely for our own interpretation.

Here is how it would look

             <<"tweet_responses_bucket">>

           ----------------------------

             |----------------------------------------|

           |   <<20337776197:128>>                  |

           |----------------------------------------|

           |   Links                                |

           |                                        |

           | {{<<"tweet_responses_subkeys_bucket">>,| 

           |  <<20337776197:128,0:8>>},             |

           |  <<"tweet_response">>},                |

           | {{<<"tweet_responses_subkeys_bucket">>,| 

           |  <<20337776197:128,1:8>>},             |

           |  <<"tweet_response">>},                |

           |  ....                                  |

           |----------------------------------------|

           |   Value                                |

           |                                        |

           |----------------------------------------|

               <<"tweet_responses_subkeys_bucket">>

           ------------------------------------

             |----------------------------------------|

           |   <<20337776197:128,0:8>>              |

           |----------------------------------------|

           |   Links                                |

           |                                        |

           |{{<<20339861590:128>>,<<18035803:128>>},|

           |  <<"reply">>},                         |

           |  ....                                  |

           |----------------------------------------|

           |   Value                                |

           |                                        |

           |----------------------------------------|

             |----------------------------------------|

           |   <<20337776197:128,1:8>>              |

           |----------------------------------------|

           |   Links                                |

           |                                        |

           |{{<<20337857101:128>>,<<82294968:128>>},|

           |  <<"reply">>},                         |

           |  ....                                  |

           |----------------------------------------|

           |   Value                                |

           |                                        |

           |----------------------------------------|

      Code

And now here is the piece of code this does the extraction of the popular replies. The function gives a sorted list of {TweetId, AuthorId} tuples which are then looked up and served.

https://gist.github.com/510070

Hopefully the code is self explanatory. Of interest is the make_local_fun which creates a function reference which can be passed over to a remote node, without the remote node having a copy of this compiled code in its path.

At inagist.com we have been using Riak and yes we are loving it. We moved away from Cassandra after it started taxing our limited resources. The nice thing about Cassandra was the data model. Super columns allowed us to store metadata for a resource as needed. For example the retweets and replies of a tweet were stored in their own super columns associated with a tweet and we could pull it out as needed. Concurrency issues were also not a bother. We could do simultaneous updates to columns and super columns and not worry about data consistency issues. This is seriously tricky when maintaining tweet statistics. Popular tweets keep getting retweeted and replied to concurrently by many people. 

When looking for alternatives Riak was our first choice primarily because of it being in Erlang and since it had a map-reduce option which looked seriously promising. The ability to have a choice of backends was another compelling factor. Here are some of the interesting stuff that we have worked out in using Riak.

Using the Data model to our advantage 

At the heart of Riak everything is a key-value. All metadata is associated with the value and has to be read and updated as a single unit. The most interesting metadata is of-course the Links that you store along with a keys value. Interesting because Riak's map reduce has an extra option called link walking. This allows you to filter the links on a document by tag or bucket and feed the linked documents to the next phase. Infact Riak's map-reduce allows you to have any combination of link, map, reduce options to process your data. And yes these are optional too. So infact you can have a link-reduce, link-link-reduce or link-link type queries too. 

Why is this interesting? It allows us to store metadata on a seperate resource and link it to the main resource. Meaning we could have say 10 buckets storing the ids of the retweets as Links and the main resource has a link to these 10 buckets. You could parse through this list of with a link-link query. This reduces the contention on one resource for updates, parallelizes the read and spreads it across the cluster. We store replies and retweet details of a tweet in this model.

A link has three attributes Bucket, Key and Tag. Its supposed to refer to a Bucket and Key if you intend to further get data out of the linked document. But if you know what you are upto this allows to do for some serious extra data management. We currently store some tweet meta-data in Bucket and Key with a well known tag. When we later want to query for all replies to a

 tweet we do a link-link-reduce on well known tags and get the replies or retweets out. I'm not getting into specifics but it should give you the idea.

Interfacing with Riak

Most references point to using Riak via the HTTP interface or via the protocol buffers client. Great if you are working from a non Erlang environment.

We currently use the built in client for Riak over the protocol buffers client. With the main processing being in Erlang, and being a distributed app at that, there was no point in going thru extra layers to get into Riak. This also gives us some interesting options, like for example the "Your Friends" tab on the conversations page that you see once you log into inagist.com. This does a link-link-reduce-reduce where an extra reduce talks to remote erlang process for the logged in user to filter out only replies from his followers. See it in action on a popular tweet like this one from @BarackObama. Mind you the "Your Friends" feature will work only after your account is enabled on inagist.com, but the Popular tab works for anyone and pulls out only the replies which are of interest.

Storage options

And yes the back-end, currently we run the innostore back-end based on Embedded Innodb. This kind of makes Riak a distribution layer over a trusted storage layer. Of the back-ends available this has worked best for us giving a consistent performance along with being reasonable on the resource usage. But the biggest factor here is that it gives an option to plugin what you want like the trial we did with Tokyo Cabinet.

Our biggest bottle neck now is the disk space, we keep pruning the data set at a failry fast pace, roughly one week of data is all that we hold. We get a little above 5 million tweets a day from the twitter pipe and we keep cleaning out as the disks fill up.