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.

 

Filed under  //  erlang   filtering   inagist   real time search   riak   riak search    search   twitter  
Posted by Jebu Ittiachen 

Link-Map-Reduce in Riak an example from inagist.com

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.

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.

Feel free to comment on anything I have overlooked or could be done better :)

Filed under  //  code   erlang   map-reduce   riak  
Posted by Jebu Ittiachen 

Riak at inagist.com

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.

A big thank you to the guys at Basho for Riak, its seriously awesome.

Filed under  //  erlang   map-reduce   nosql   riak  
Posted by Jebu Ittiachen