RubyScale

Mi Casa Verde. You can read the specifications of the devices for yourself but I like them because they are inexpensive, have low-power requirements, are capable of interfacing with a number of different kind of home automation systems, and most importantly, have a documented JSON/XML API. Coincidentally, they also have a developer special program and lack a ruby gem, so another rainy day project is born! The operating system these devices run is called MiOS which is essentially Linux/ARM and lots of glue. Mi Casa Verde operates a free VPN service that allows you to operate them remotely, and there are a number of free/paid smartphone apps available (though none struck me as particularly special). All of the interactions with MiOS work on a job queue basis. When a job is submitted, it's status must be polled for success/failure. The Ruby library embraces this and allows jobs to be run synchronously or asynchronously. Some snippets from the github page:

mios = MiOS::Interface.new('http://192.168.15.1:3480') switch = mios.devices[0] switch.off! { |obj| puts "The #{obj.name} is now off" } puts "This will get printed once the switch is off and the block has been executed" switch.on!(true) { |obj| puts "The #{obj.name} is now on" } puts "This will output immediately" sleep(5) # Sleep to wait for the thread to finish

I recently has a use for HTTParty in a project that potentially required multiple base_uri references at the same time. Just changing the base_uri isn't thread-safe so you can't do that so the most common way around this with HTTParty is to make an instance variable and pass that around to all of the calls to get/post/etc. I find that a little cumbersome so instead I made use of dynamic classes to hide all of that nonsense. In a stripped down example, an HTTParty class looks somewhat like this:

class Foo include HTTParty base_uri 'http://example.com' def bar self.class.get '/some/resource' end end

  But rewriting it like so allows the same class usage but with variable base_uris:

class Foo class << self def new(uri, *args) Class.new(AbstractFoo) { |klass| klass.base_uri(uri) }.new(*args) end end class AbstractFoo include HTTParty def bar self.class.get '/some/resource' end end end foo1 = Foo.new('http://example.com') foo2 = Foo.new('http://otherexample.com')

I recently had a conversation in #cassandra about the Data Model that I thought might be useful to try to distill into a few lines. These few lines ignore all of the implementation details to make it work in practice but it gives you the starting point. It looks a little something like this ...

Keyspace - What method should I use to turn my application key into a Row Key?

Row Key - Which Node is my value on?

Column Family - Which file on the node is my value in?

Column Name - Which piece of the file on the node contains my value?

Column Value - My value!

Since the columns are already sorted in the file, getting a slice of them is very efficient, and this is what makes Cassandra a column-oriented database. It's worth noting that with most Cassandra clients, the Row Key and the Column Family are almost always conceptually swapped because it's more common to access multiple keys in the same column family than to access multiple column families with the same key. Given the above, let's revisit some of the other concepts that are required to make it work in practice.

Partitioner

The partitioner determines how your application keys get turned into Row Keys. RandomPartitioner (which should be called HashedPartitioner in my opinion) takes an MD5 of the key and uses that as the Row Key and ByteOrderedPartitioner uses the key unmodified.

Replication Factor and Replica Placement Strategy

ReplicationFactor determines how many nodes get a copy of a particular key. There is no master/slave thing going on here, it's just how many nodes get a copy of a key. The Strategy determines which exact nodes get a copy of your key based on network topology or ring placement.

Consistency Levels

Most relevant when the ReplicationFactor > 1, the ConsistencyLevel determines how many nodes have to successfully record a write or agree on the value of a read for the operation to be considered successful. There are tables of consistency levels and what exactly they mean in multiple places so I won't repeat them here.

Memtables, Commitlog, SSTables, and Compaction

As writes come in to a ColumnFamily, they are simultaneously (don't argue) stored in memory in a memtable and written out to disk in a commitlog. The memtable is periodically written out to disk in in column-order as an sstable for efficient slicing later. Over time, a column could exist in a memtable and/or multiple sstables at once so a timestamp is used to determine which one to use, highest one wins. The timestamp is supplied by the client and can be any integer. Over time, you get a number of sstables and a process called compaction combines them all and throws away the out of date stuff to save space. Back to the commitlog, that's there so that if the node explodes while some things are only in a memtable, they get replayed when the node starts back up so no data is lost (durability). Since the commitlog is serial write-only, if you put it on a separate disk, you can accept writes really fast.

Miscellaneous

You can ask any node anything, they all act as proxies for one another.

If your ReplicationFactor > 1, then ReadRepair will propagate the most recent column value to the nodes responsible or it.

There are various operations in nodetool that let you move nodes, add nodes, clean out old data/keys, manually compact things, repair everything at once, get statistics about each node, etc.

Gossip is used to propagate node status and information to other nodes.

Cassandra Query Language (CQL)

Cassandra originally went with a Thrift RPC-based API as a way to provide a common denominator that more idiomatic clients could build upon independently. However, this worked poorly in practice: raw Thrift is too low-level to use productively, and keeping pace with new API methods to support (for example) indexes in 0.7 or distributed counters in 0.8 is too much for many maintainers. CQL, the Cassandra Query Language, addresses this by pushing all implementation details to the server; all the client has to know for any operation is how to interpret “resultset” objects. So adding a feature like counters just requires teaching the CQL parser to understand “column + N” notation; no client-side changes are necessary. (CQL Specification: github.com/apache/cassandra/blob/trunk/doc/cql/CQL.textile)

Cassandra-CQL For Ruby

While working with completely virtualized systems and architectures is fun, for me it's never been quite as much fun as doing things in the Real World™. A few years ago I found Phidgets Inc. Phidgets are a set of "plug and play" building blocks for low cost USB sensing and control from your PC. From their website:

All the USB complexity is taken care of by our robust API. Applications can be developed quickly by programmers using their favorite language: C/C++, C#, Cocoa, Delphi, Flash AS3, Flex AS3, Java, LabVIEW, MATLAB, Max/MSP, MRS, Python, REALBasic, Visual Basic.NET, Visual Basic 6.0, Visual Basic for Applications, Visual Basic Script, and Visual C/C++/Borland.NET.

Unfortunately, there is one language that is absent from this list: Ruby.

Several years ago, there was an effort started by Craig DeHaan and later picked up by Kit Plummer which worked pretty well but the callbacks seemed a little fragile and I wanted the interface to be a little more idiomatically Ruby. Also, using FFI means it works on multiple Ruby VMs without issue, MRI, JRuby, etc. Like the previous libraries, there aren't many devices supported yet but that's just a matter of having them. There are traditional and idiomatic examples in the repository and it can be installed with everybody's favorite command: gem install phidgets-ffi.

editors note: as of 7 Mar 2012, this info is still current and correct.

One of the most common use cases for Cassandra is tracking time-series data. Server log files, usage, sensor data, SIP packets, stuff that changes over time. For the most part this is a straight forward process but given that Cassandra has real-world limitations on how much data can or should be in a row, there are a few details to consider. Basic Inserts and Queries The most basic and intuitive way to go about storing time-series data is to use a column family that has TimeUUID columns (or Long if you know that no two entries willhappen at the same timestamp), use the name of the thing you are monitoring as the row_key (server1-load for example), column_name as the timestamp, and the column_value would be the actual value of the thing (0.75 for example):

Inserting data -- {:key => 'server1-load', :column_name => TimeUUID(now), :column_value => 0.75}

Using this method, one uses a column_slice to get the data in question:

Load at Time X --  {:key => 'server1-load', :start => TimeUUID(X), :count => 1}

Load between X and Y -- {:key => 'server1-load', :start => TimeUUID(X), :end => TimeUUID(Y)}

This works well enough for a while, but over time, this row will get very large. If you are storing sensor data that updates hundreds of times per second, that row will quickly become gigantic and unusable. The answer to that is to shard the data up in some way. To accomplish this, the application has to have a little more intelligence about how to store and query the information. For this example, we'll pick a day as our shard interval (details on picking the right shard interval later). The only change we make when we insert our data is to add a day to the row-key:

Inserting data -- {:key => 'server1-load-20110306', :column_name => TimeUUID(now), :column_value => 0.75}

Using this method, one still uses a column slice, but you have to then also specify a different row_key depending on what you are querying:

Load at Time X -- {:key => 'server1-load-<X.strftime>', :start => TimeUUID(X), :count => 1}

Load between Time X and Y (if X and Y are on the same day) -- {:key => 'server1-load-<X.strftime>', :start => TimeUUID(X), :end => TimeUUID(Y)}

Now what to do if X and Y are not on the same day? No worries! You can use a multi-get to fetch more than one key at a time (or issue parallel gets for maximum performance). If your X and Y span two days, you just need to generate keys for those two days and issue them in a multiget:

Load between Time X and Y -- {:key => ['server1-load-<X.strftime>', 'server1-load-<Y.strftime>'], :start => TimeUUID(X), :end => TimeUUID(Y)}

Then in your application, you will need to aggregate/concatenate/iterate those two rows however you see fit. If your data spans 3 or more days, you'll need to also generate every key in between. Don't be tempted to use the Order-Preserving Partitioner here, it won't save you that much typing and it'll will make managing your cluster much more difficult. Calculating Shard Size Now on the topic of determining your shard interval .. that's a complicated topic that is often application dependent but the single biggest issue is to make sure your rows don't get too big. The better you are at the ops side of Cassandra, the larger you can let your rows get but if I have to tell you that, you should keep them small. A quick ballpark formula for determining shard size is as follows (yes rcoli, it ignores overhead):

shard_size_in_seconds / update_frequency * avg_data_size_in_bytes == row_size_in_bytes

Set your shard size so that the row_size doesn't get much larger than 10MB (this number can move around for many reasons but I'd consider it safe). For example, if you are storing hits on a website that gets 10 hits/sec and each entry is about 200B, then we have:

Daily -- 86400 / (1 / 10) * 200 = 172800000 (165MB)

Hourly -- 3600 / (1 / 10) * 200 = 7200000 (6.9MB)

Since most developers who have used databases are familiar with traditional RDBMS systems, I'll explain Cassandra SuperColumns as succinctly as possible using some tables.

The first is an example of a row from a column_family.

select * from user_1;

column_namecolumn_value name kelley reynolds favorite_drink hard cider

As you can see, this shows some name/value pairs, and those values are easy to get quickly. Now a supercolumn, on the other hand, is usually illustrated like so:

select * from users;

supercolumn_namecolumn_namecolumn_value user_1 name kelley reynolds user_1 favorite_drink hard cider user_2 name kevin way user_2 favorite_drink port ellen

But this would be WRONG! It would actually be illustrated more like this:

select * from users;

supercolumn_name subcolumns user_1 {name:'kelley reynolds',favorite_drink:'hard cider'} user_2 {name:'kevin way',favorite_drink:'port ellen'}

And now you should be thinking to yourself, "Self, that's silly! I'd never do it that way. If I wanted to access just the favorite drink of one of those people, I'd have to read the entire thing, deserialize it, then read just the row I was after. I can't read just one subcolumn with it done that way!"

RaphaelJS is a fantastic vector library, but as it is a vector library, its not meant for general markup. Trying to include dynamic text wrapping or videos or any HTML really can be a pain and the only real method for doing that is to position a div in the right spot and account for it, taking into account page load time, scroll bars, resizing, etc. Well, why not use a little jQuery to wrap up a nice little object to completely handle that for you?

// Takes a raphaeljs object, some options, and some container attributes function Infobox(r, options, attrs) { options = options || {}; attrs = attrs || {}; this.paper = r; this.x = options.x || 0; this.y = options.y || 0; this.width = options.width || 250; this.height = options.height || 245; this.rounding = options.rounding || 10; this.container = this.paper.rect(this.x, this.y, this.width, this.height, this.rounding).attr(attrs); var container_id = this.container.node.parentNode.parentNode.id; container_id = container_id || this.container.node.parentNode.parentNode.parentNode.id; this.raph_container = jQuery('#' + container_id); this.div = jQuery('').insertAfter(this.raph_container); jQuery(document).bind('ready', this, function(event) { event.data.update(); }); jQuery(window).bind('resize', this, function(event) { event.data.update(); }); } Infobox.prototype.update = function() { var offset = this.raph_container.offset(); this.div.css({ 'top': (this.y + (this.rounding) + (offset.top)) + 'px', 'left': (this.x + (this.rounding) + (offset.left)) + 'px', 'height': (this.height - (this.rounding*2) + 'px'), 'width': (this.width - (this.rounding*2) + 'px') }); }

    And that's pretty much it. To update the content, you just use the .html() method like any jQuery operation. You can even couple events to the content and the infobox if you want to like so:

Infobox.prototype.show = function() { this.container.animate({opacity: 1}, 400, ">"); this.div.fadeIn(200); } Infobox.prototype.hide = function() { this.container.animate({opacity: 0}, 400, ">"); this.div.fadeOut(200); }

And now you can show/hide the content and the SVG bounding box with whatever effects make you the happiest.

I'm sure you were expecting another post with a brief topical analysis of why Facebook chose HBase instead of Cassandra with some similarly topical and mostly incorrect summary of how those two systems work and why one is better than the other one.

Sorry to disappoint, but rather than pour more fuel on what is already a pointless fire, I'm going to suggest instead that you look at what your application and your business requires for success and can operationally support. That might mean that you need transactional rollback so Cassandra won't work straight up, or that you already have HDFS for other jobs so HBase is easier to support than adding on an entirely new technology.

If you read the last post, Understanding the Cassandra Data Model from a SQL Perspective, then you already have a decent handle on what a column-oriented key/value store is. The next thing that tends to confuse new users of Cassandra is the notion of Consistency Level and Replication Factor. This post doesn't really have a lot to do with SQL in particular, it's more of a discussion about how tables would be replicated across multiple servers and how you read/write from those tables.

1. Replication Factor

In the previous post, we described how every single Cassandra Row would be a separate table in SQL with an arbitrary number of rows. The Replication Factor (aka 'RF') determines how many nodes have a copy of that table. If you have 10 nodes and your RF=3, then every single table will exist on 3 nodes. Table users-91b56770-bf40-11df-969c-b2b36bca998e might exist on nodes 2 through 4 and users-28b56770-b410-15ef-968c-b2c36d511e78 might exist on nodes 6 through 8.

Can the RF be any number? Anything one or higher works but make sure it's not higher than the actual number of nodes you have. This may be fixed by now but it used to be handled ungracefully. In general it should also be an odd number >=3 so that quorum is a useful consistency level (for more details on this, keep reading).

How do I know which node to query to get my data if it could be anywhere? Do I have to know where to write it? Each node acts as a storage proxy for every other node. If you query node 5 for data that is on nodes 8-10, it knows to ask one or more of those nodes for the answer. This is also true of writes.

What if a node goes down? How do I synchronize them? What if they have different values? This will be covered in the next section, Consistency Levels.

2. Consistency Level

Consistency in Cassandra is Eventual, which is to say that the N nodes responsible for a particular table could have a slightly different opinion of what it's contents are. That's ok! When reading from or writing to a Cassandra cluster, you can tune the tolerance for that sort of tomfoolery by reading or writing at a varying Consistency Level (aka 'CL'). Remember that each node acts as a proxy for every other node so you don't have to worry about which node you are interacting with (unless you are really trying to optimize network activity but in general, this is unnecessary).

CL::ALL will not return success unless it has successfully written the same value to every node responsible for holding whatever you are writing. Using CL::ALL, you can guarantee that whatever you are writing is the same on all nodes when success is returned. There is of course a performance penalty for this, but in return you get very strong Consistency. Reading at CL::ALL functions in a similar way .. if one of the nodes does not respond, it will return an error since it cannot guarantee that all of the nodes are consistent for the value in question.

CL::QUORUM only requires a majority of the nodes responsible for a table to respond instead of all of them. This allows for one or more nodes to be down or otherwise unavailable and still have your read/write return. If your RF=3, then only 2 nodes responsible for a table need to be online in order to reliably manipulate the data in that table. Note that if RF=1 or RF=2 then there is no meaningful difference between CL::ALL and CL::QUORUM so if you want the benefits of CL::QUORUM, make sure your RF>=3.

CL::ONE requires that only one of the nodes responsible for a table respond. This makes reads and writes fast, but it also means that depending on what else is reading and writing, it's possible that they could briefly give conflicting answers. This is a fine tradeoff for speed in many applications, but not all. For example, recording votes or website hits where the outside possibility of a few going missing on machine failure is probably fine. Recording a financial transaction of some sort, probably not so much.

CL::ANY is only used for writing, not reading. CL::ANY means that as soon as a write is received by any node, the call returns success. This occurs when your client might be connecting to node 5 but the nodes responsible for it are 6-8. The difference between CL::ONE and CL::ANY is that with CL::ANY, as soon as node 5 receives the write, it returns success (but nodes 6-8 could be down or whatever). CL::ONE means that if you write to node 5, either 6, 7, or 8 have to return success before node 5 returns success.

OK .. so if I use CL::ALL and it fails, does that mean that my write failed? Not necessarily! It may have succeeded on two nodes and failed on the third which means that eventually it will be propagated to the third, but the required guarantee was not met.

Uh .. ok. So if that happens, how do I know that any of my writes succeeded? You don't! At least not yet. The error doesn't indicate percentage of success, just failure. Not great, I know.

I have multiple datacenters. How do I tell Cassandra to put some data in each? There are various strategies and snitches that tell Cassandra where to put things and they are constantly in flux. You can also write your own if you have specific requirements. For more information, see the Cassandra Wiki.

What happens if I write at CL::ANY and that node explodes before it can send data to where it belongs? Congratulations, you just lost data. If you can't tolerate that, don't use CL::ANY.

What happens if all 3 nodes are up, but have different values for something for some reason and I read at CL::ALL. Does it give me the finger or what? Cassandra performs 'read repair'. That is, upon reading, if you are using a CL that requires more than one node and they disagree on the value, Cassandra will compare the values and use whichever one has the latest timestamp. It will also write that value back to the node which had an outdated value hence 'repairing it'.

If you come from the SQL world, sometimes it can be difficult to understand the Cassandra Data Model and all that it implies in terms of management and scalability. For this post, we're going to go backwards. Instead of writing an example application using Cassandra to understand it, I'll describe implementing Cassandra on a traditional SQL database and what that would look like.

1. Mapping a Column Family to SQL Tables

In Cassandra, a Column Family has any number of rows, and each row has N column names and values. For this example, let's assume that in Cassandra we have a Users Column Family with uuids as the row key and column name/value pairs as attributes such as username, password, email, etc. If we had 10000 such users, then in SQL we'd have 10000 tables that looked like the following where 91b56770-bf40-11df-969c-b2b36bca998e was the actual uuid of a user:

CREATE TABLE users-91b56770-bf40-11df-969c-b2b36bca998e (column_name varchar, column_value varchar) UNIQUE column_name PRIMARY KEY column_name;

INSERT INTO users-91b56770-bf40-11df-969c-b2b36bca998e VALUES ('username', 'joe'), ('password', 'weak'), ('email', '[email protected]');

10000 tables! That's right, 10000 tables. In SQL, that many tables would be absurd but Cassandra is designed to scale that way and lots of rows are encouraged as it helps load balancing and scalability. In fact, millions and billions of rows are the norm. You can think of Cassandra as a system that distributes these tables over many nodes with some consistency and replication guarantees.

You said N rows, but then said a table is on a single node. You don't really mean N do you? No, not really. A row has to fit on a single node just like a table has to fit on a single machine for most SQL implementations. For older versions of Cassandra, a row also had to fit in RAM.

It looks like I can only query one user at a time and I already have to know his UUID. Is that true? In SQL terms, Cassandra can perform table ranges and row slices on table ranges. In Cassandra terms, you can get a slice of keys and a slice of columns within those keys. This is what is meant by 'column-oriented key-value store'. One thing you don't get to do is join. Cassandra does not join. Cassandra does not join. If you want to join, you do it in your application.

What is a Column Family for then? Just a table prefix? A Column Family has a number of settings that go with it that alter it's behavior. There are cache settings for the keys (the UUIDs in this example), cache settings for the entire rows (the entire table in this example), and most importantly, sorting. In Cassandra there is no OFFSET, only LIMIT and the equivalent of BETWEEN. In this example, the column names are just strings but they could also be integers or timestamps and they are always stored in sort order. One Column Family might have timestamp-sorted data where you query things by time slice and another might be address book data where you query things in alphabetical order. The only sorting you get to do after the fact is reversing a particular slice. 

What if I want to sort by column value and by column name? Then you have two Column Families, one sorted by whatever datatype your column names are, and one sorted by whatever your column values are.

So if I wanted to find the user who had a particular email address, how would I do that? I don't know what table to query! Exactly! If you want to answer that question, you'll have to make another Column Family that keys on e-mail address and points to the UUID of the user that has it, then you can query the relevant users table. Think about it this way .. in SQL, you can CREATE INDEX for things you want to query and in Cassandra, you have to manually maintain that index. (Secondary indexes are present in 0.7+, but I'll cover that another time.)

In SQL, I can still do a full table scan to pull out the information I'm looking for. Why can't Cassandra do that? You can iterate with Cassandra and functionally get the same result, but I wouldn't if I were you.

But if I have two Column Families with the same data sorted differently, that's denormalized. I heard that was bad. Denormalization is the norm with Cassandra. This is a complicated topic but in short, denormalization is a requirement for linear horizontal scaling and many people were already doing this in their SQL databases to avoid joins anyway.

Wait, there is no OFFSET? How do I paginate with cassandra? Awkwardly. You have to start at the beginning, get N items, then on the next page you'd have to start at the last of the previous items and get N+1, throwing away the first one, etc. Alternately, you can create another Column Family which caches pages but in general, pagination is not what Cassandra was designed for.

  2. Mapping a Super Column Family to SQL tables

A Super Column Family is exactly like a Column Family, except you get one more column in your table. For this example, we'll make an address book.

  CREATE TABLE addressbooks-91b56770-bf40-11df-969c-b2b36bca998e (super_column_name, column_name varchar, column_value varchar) UNIQUE (super_column_name, column_name) PRIMARY KEY super_column_name;

INSERT INTO addressbooks-91b56770-bf40-11df-969c-b2b36bca998e VALUES ('bob', 'street', '1313 Mockingbird Lane'), ('bob', 'city', 'Chicago'), ('bob', 'state', 'IL');

INSERT INTO addressbooks-91b56770-bf40-11df-969c-b2b36bca998e VALUES ('alice', 'street', '123 Foo St.'), ('alice', 'city', 'Kona'), ('alice', 'state', 'HI');

Do the column_names have to be the same for each Super Column? No.

Can the super_column_name and column_name be sorted differently? Yes.

Does the entire table still  have to fit on one node or does it split out the Super Columns? The entire row still has to fit on one node.

Is there a limit to how many Columns can be in a Super Column? Other than the space limitation already mentioned, theoretically, no. There is one gigantic caveat: Accessing a single Column in a Column Family is efficient and fast. Accessing a single Column in a Super Column ('city' or 'state' in this case) requires deserialization of the entire Super Column. For this example, this means that to get alice -> state, Cassandra has to load everything in the alice super column. For an addressbook, this doesn't really matter but if you had 1000 columns and a read-heavy load, it could heavily impact performance since you'd have to deserialize all 1000 columns just to get one of them.

It sounds like Super Column Families can be dangerous if I'm not careful. Exactly! They were designed specifically for inverted indexing. If you aren't doing that or don't know what that is, don't use them.

  3. Transactions and/or Rollbacks

Cassandra does neither transactions nor rollbacks. I you want those, some other thing such as Cages/ZooKeeper has to do that.

Next Post

Our recently-released Cassandra storage plugin for paperclip was lacking one thing: the ability to specify read/write consistency levels. Well, now you can just by adding the following options to your has_attachment specification (these are the defaults, btw).

:read_consistency => Cassandra::Consistency::ONE :write_consistency => Cassandra::Consistency::QUORUM

These have already been sent to the port maintainers so they should find their way into the ports collection soon enough but if you can't wait, here are some scripts for pfstat (pfstatd in particular) and cassandra.

/usr/local/etc/rc.d/pfstatd

#!/bin/sh # PROVIDE: pfstatd # REQUIRE: LOGIN # BEFORE: securelevel # KEYWORD: shutdown # Add the following lines to /etc/rc.conf to enable `pfstatd': # # pfstatd_enable="YES" # pfstatd_flags="" # # See pfstat(1) for pfstatd_flags # . /etc/rc.subr name="pfstatd" rcvar=`set_rcvar` command="/usr/local/bin/pfstatd" # read configuration and set defaults load_rc_config "$name" : ${pfstatd_enable="NO"} run_rc_command "$1"

  /usr/local/etc/rc.d/cassandra