In-A-Gist Blog

Why Erlang?

I often get these weird looks whenever I mention that inagist is written in Erlang. So here is some of the key areas where Erlang is a winner for us.

What we do

At inagist we try to summarize a real-time stream in real-time. Currently we work on top of the Twitter stream api. By summarizing I mean filter in real time the popular tweets in a stream and group tweets based on trends. See it in action at justinbieber.inagist.com, libya.inagist.com (try with chrome / safari for best results since it uses websockets). We do this summary on a stream which could be combined in any number of ways. ie; a users own stream (my stream), a keyword based search stream (libya.inagist.com), keyword + geo-location based tweet stream (sxsw.inagist.com).

Lightweight Processes

The key differentiator here is the real-time nature of how we summarize the tweet stream. Instead of possibly persisting each tweet in the stream and running off-line analytics we maintain limited caches for each user and keep popping in and out each tweet as it gains popularity or when key words are repeated in the stream for trend detection. Here is where a key aspect of Erlang fits in. Each of the stream consumer is modelled as an Erlang process, being light weight and isolated. It essentially is a proxy for each user provisioned. It receives tweets from the stream, manipulates the caches and responds to api queries for serving data. Each of these stream consumers are gen_server implementations, tied in a supervisor chain. In case one of the consumers goes down the supervisor brings it back up with no impact on the rest of the user base. 

Messages

So how do we couple this to the stream of incoming data, with messages. Each tweet is delivered to the consumers as messages from the stream api client. Each tweet consumer is part of a process group tree spanning across machines. The moment a tweet is received from the n/w it is spawned off as a seperate process which json decodes the tweet and send messages to the distribution tree. The message trickles down to the consumer process which does its job of cache updations. Being asynchronous messages the client is not concerned about how many consumers there are to a tweet. If a consumer process is interested in a tweet it can consume it. Being a decentralized delivery it scales when load of incoming tweets goes up or if the number of consumers increase.

Distributed Erlang

As the number of consumers increase another key aspect of Erlang comes into play. Distribution across machines. Consumer processes by design are known only by a process id. Erlang works with local process ids the same way as it does with distributed processes. As long as a consumer is part of the distribution tree the tweet will be delivered to the process. This helps us in easily scaling out. Individual machines could fail independently without affecting the cluster as a whole only users provisioned on that machine are offline for the time the machine is offline.

Real Time Delivery

To be as real-time as possible we prefer to deliver over websockets to connected clients. Messages come into play here again. Each stream consumer generates messages as its caches figure out a trending tweet or trend. Web socket clients tap into this message stream, convert them to JSON and deliver to the client. Our chrome browser plugin is a websocket client which utilizes this delivery model. The extension notifies via desktop popups when ever a trend is detected or a tweet gains popularity above a certain level. We also bring in a different angle to real-time search with the extension. When a trend is detected the extension automatically starts searching for prominent tweets with those detected trends. 

Streaming Search

I have written previously about how we use Riak for search and storage. In addition we have built some custom stuff which enables streaming search. Whenever we index documents in Riak we also send out the indexing data as messages to the search infrastructure. We also send this index terms as soon as a tweet is received on a seperate index. Here we have process waiting for <index, field, value> tuples to match against and notifies waiting processes of a document matching the search criteria. We currently support ==, and, or, >, <, >=, =< operators so we can detect any tweet containing sxsw (==), justin bieber (and), documents containing a text or lat long within a bounding box etc. Stream consumers use this to get real time filtered tweets from the stream. Also the chrome plugin taps into this search stream to notify a user whenever a tweet matches a detected trend or an explicit search query. This is really powerful since we now automatically figure out what a users interest topics are by way of trends and we can let the user know whenever there is something matching this interest topic in real time. This whole streaming search works with such low over head because of the nature of the the message based architecture that we can stream this all the way upto the browser typically under a second or two. You can see it in work when you click on the "Live Stream" heading on pages like these justinbieber.inagist.com.

I have just given a high level view of where Erlang acts as a differentiator but it should give some insight into why we do Erlang all the way. Drop in a comment or @jebui and will be happy to give more information if needed.

PS: I have not heard any of Justin Bieber's or Lady Gaga's albums they just happen to have a very active tweet stream.

Posted by Jebu Ittiachen 

Searching with RiakSearch

I had in my previous post mentioned what we do with the search functionality at inagist.com. This post will look into the technical details behind the implementation. We use Riak as our storage layer, RiakSearch was a natural add on to it. I will try to detail my understanding of how riak search works and how we use it.

At its heart RiakSearch is an inverted index of terms to document id's. The inverted index maintains an ordered set of document id's, the merge_index backend which stores this index, splits this across various files. Specifically the backend has buffers which maintain the index in ETS as well as files, and segments which are files using a custom format to store ordered keys associated with an index, field name, field value tuple. Segments store metadata information regarding file offsets for key lookup and are loaded into ETS at startup for faster access. Additionally bloom filters speed up lookup in each offset. Buffers are periodicaly merged into segments, and segments once created are not updated except for the merge of segments into a single segment. Very much like the BitCask store for Riak. All this happens at the vnode level and riak core sits on top of all this and distributes the operations across vnodes. Index name, field name and field value are used to determine the hash for mapping to a vnode. At indexing time a document is split into postings which have index name, field name, field value mapping to document id and a bunch of properties. These are batched and send to vnodes responsible for each hash in parallel. 

Queries are broken into a set of logical operations which combine each individual matching term and brings up a final list of matching documents which are sorted and ranked. A query like "tweet:facebook email" is broken into something like "tweet has facebook and email". This translates to a logical and of docs having tweet:facebook and tweet:email, these operations are then send to the vnodes to stream the doc-id's matching this operation. The doc id's are then merged in order, via a merge sort since the keys are already sorted. This results in a final list of doc id's matching the query and the properties for each doc. The results are sorted based on these properties and finally returned. The properties have a term frequency for each doc and a pre-plan operation give the document frequencies for each term allowing to sort the docs based on term frequency and inverse document frequency.

Now that was a whirl-wind simplified wrap up of my reading of the search code. To note here is that its a very performance aware implementation for indexing and simple queries. Queries with low cardinality terms could throw away your search times, there is something in the works for this specific issue with inline fields. Also queries which could return millions of rows are also possible memory busters, even if you give options to limit the number of results these are operations which happen after the full results of the operation are in memory. Both of these re-iterates the fact that this is as the name implies "RiakSearch", an addon to make your life easier when working with Riak. The implementation is tailored for mating with the map-reduce operations for riak and that is readily exposed via all the interfaces to riak search.

How do we use it?

The search box on inagist.com is directly wired into riaksearch. To prevent our queries from leading to memory exhaustion we do a couple of tricks. As previously mentioned our backend is fully in Erlang and we directly talk to Riak in Erlang. We directly call the search_fold on the riak_search_client from our code and break the search operation when we have enough results. Our keys, the tweet id's are stored as negative numbers so that the sort ordering of the keys means we get the first n docs ordered in a latest first manner. We then rank them in this limited set.

The next place we use search is for threading conversations on the tweet detail page. I had in an earlier blog post mentioned how we did that with links and and link-map-reduce operations. With search we just index the replies against the tweet it is in reply to and bring it back in from search on the reply to field. This is better since link updates modifies the whole document un-necessarily, where we only needed the meta-data on the tweet to be updated.

Another place we plugin search is to run our clean-up operation. We index the tweet timestamps to the minute granularity and cleanup tweets older than a certain time period. Getting to older tweets without search would have meant we maintain the ids separately to get a handle on which id's to flush out.

While you can choose to have riaksearch index all documents that are stored in your riak cluster via a pre-commit hook, we decided to trigger the indexing via our own calls into riak search. Two reasons to this, we found the pre-commit hook fail a couple of time with a timeout under heavy load, also our indexing needs meant we index the text in a tweet at a point when the tweet was determined to be indexable by the app and not at the point of insert into the back end store. Params like the time stamp however are indexable at insert time.

Final Thoughts

RiakSearch perfectly complements Riak key value store. It frees you from having to access documents by id alone and managing your data is simpler. The fact that it works well with existing java code for text analysis is also worth mentioning. Its still in beta so I guess things are only going to get better from here.

Posted by Jebu Ittiachen