How to append data to a Riak key under a heavily distributed environment?

Question

Using Riak I want to append data sequentially in a way that I can obtain all of the data I appended from time to time. Think of logs, if I pick incremented log rows and transfer them to riak, at some point I want to reconstitute all what I have appended.

I thought of doing this by creating a new bucket for that purpose, then add keys defined by a sequential number or datetime stamp, and add the content to it, then use the list keys API and reconstitute the data I need. The problem with that is that the list key API is not efficient and production recommended. What I like about this approach is that the data has no concurrency write problems ( no locks/etc ) since all keys are independent.

The other approach is to use a single key, open it and append to it, but I am very concerned by concurrency/locking issues. This action would be performed under a distributed environment and would be certainly a bad choice

Question: any other ways to do that in Riak ? Any append-mode to a key ?

Answer 1

While writing lots of small records to a bucket is easy and efficient at write time, the penaly you pay is that it instead gets expensive when you try to read the values as you most likely will not know the keys. If you need to look these keys up by a secondary index, key filter or even worse, by going through all the keys in the bucket (which is a very heavy operation and never recommended for a production environment), this will be considerably less efficient than retrieving the data by key, and not scale as well.

There is also no append functionality in Riak, which means that you will need to first read and then write a record in order to update it and append new log entries. Depending on how you organize and coordinate your writing, this can as you point out result in concurrent updates of the same record, which need to be considered when designing a solution.

Assuming the records you are collecting, e.g. log entries, can be treated as a set, a technique I would recommend is time-boxing. When time boxing, you aggregate data based on a time period. If we e.g. assume that we are collecting logs for a set of servers (named server in this example), we can create records having keys based on the server ID and a datetime identifier, e.g. the start of the measurement period. We don't need a full timestamp, just enough to allow us to identify the record. A record holding log entries for server3 covering the period between 14:15 and 14:20 on 2013/03/07 could be named 'server3_20130307_1415'. The following 5 minute period would then accordingly be named 'server3_20130307_1420'. If there is no data for a period, no record will be created.

This allows you to automatically know the key for a record covering a specific period, and will allow you to retrieve records based strictly on key access, which scales and performs very well. You would naturally need to adjust the time period covered by a single record depending on the amount of data you generate, as you generally want to keep the size of objects in Riak below 1-2MB. It is also worthwhile considering compressing data at the application level if each period would have a lot of data, in order to get below this recommended size.

If you wanted to be able to access larger chunks of data without having to retrieve a potentially large number of records, you can periodically aggregate records. You could e.g. read all records covering an hour and write the aggregated data to a new record named 'server3_20130307_14' that covers the entire period 14:00-15:00. As you know the keys, this is straight forward and easy to implement as a batch job.

When taking this approach, you will, as discussed previously, need to consider the possibility of concurrent writes. The best way to do this is in my opinion by allowing siblings (set 'allow_mult' to true and 'last_write_wins' to false for the bucket using bucket properties [1]). This will cause Riak to keep all versions of the record in the case of concurrent updates, and you will instead need to resolve any siblings created in your application layer upon reading a record with siblings. Although this does add a bit of complexity, it ensures you will not lose any data.

As we assumed the log entries in this case can be treated as a set, you can merge the sets of all siblings through a set union and then update the object (with the correct vector clock) in order to resolve the siblings.

[1] http://docs.basho.com/riak/latest/references/apis/http/HTTP-Set-Bucket-Properties/