More precisely, Cassandra has always provided row-level atomicity of batch mutations. This means that multiple batched writes to the same row are persisted by nodes atomically. When doing     UPDATE Users     SET login='eric22' AND password='f3g$dq!'     WHERE key='550e8400-e29b-41d4-a716-446655440000' Cassandra guarantees that the new login and password are either both persisted or none are.

However, up to Cassandra 1.0, the isolation of such an update was not guaranteed. In other words, it is possible (during a very brief moment during the update) that a read like
    SELECT login, password     FROM Users     WHERE key='550e8400-e29b-41d4-a716-446655440000' returns the new login ('eric22') but not the new password ('f3g$dq!'). This changes in Cassandra 1.1 as row-level updates are now made in isolation. Cassandra 1.1 guarantees that if you update both the login and password in the same update (for the same row key) then no concurrent read may see only a partial update.

These atomicity and isolation guarantees apply to columns written under the same physical row, i.e. that are within the same column family and share the same partition key. For atomicity, the guarantee actually extends across column families (within the same keyspace): updates for the same partition key are persisted atomically even for different column families. This is not the case however for isolation (updates to different column families are not isolated).

Note that when we say that Cassandra persists row-level writes atomically, this applies to each node of the cluster individually; Cassandra does not provide any cluster-wide rollback mechanism. In the preceding example, the guarantee is that the new login cannot be persisted without the new password being persisted too (and vice-versa). It is however possible for both to be persisted even if the client operation end up with a timeout (because not enough nodes have acknowledge the write to satisfy the requested consistency level). It is up to the client to retry a failed write in such cases.

Implementation details

Atomicity

Internally, the row-level atomicity is guaranteed mainly by the commit log. Upon reception by the coordinator, each write query is transformed into a bunch of ‘RowMutation’. Each of those RowMutation regroups all updates for a given row key (even for different column families). On every replica, each RowMutation is first serialized and written to the commit log as one mutation (individually checksummed for assessing integrity in case of failure). This ensures that on failure, that RowMutation is either replayed entirely (if it had been completely written in the commit log and isn’t corrupted) or not at all. The other part of guaranteeing the atomicity of persistence comes from the fact that a given RowMutation is applied to one and only one memtable. It follows that the RowMutation (all the updates from a client query for a given row key) can only be persisted together or not at all.

Isolation

To a large extent, the log-structured nature of Cassandra storage engine makes row-level isolation easier. Writes are applied to memtables that are then persisted as sstables which are immutable. Thus ensuring that a RowMutation is applied to the current memtable in isolation (of other writes and reads) is enough to ensure complete isolation. That is what was added to Cassandra 1.1, the application of RowMutation to memtables in isolation. Technically, we use SnapTree copy-on-write clone facilities: all the columns of a new mutation are applied to a non-visible (and thus isolated) copy of the in-memtable row they are applied to and then we atomically replace the original row with the new copy through a compare-and-set.

The best of both worlds

In a static-column storage engine, each row must reserve space for every column

In a sparse-column engine, space is only used by columns present in each row

Clustering, composite keys, and more

Under the hood and historical notes

Raw tweet data

Logical representation of the denormalized timeline rows

Physical representation of the denormalized timeline rows

Physical representation of the denormalized timeline rows, WITH COMPACT STORAGE

The main design goals for the Cassandra File System were to first, simplify the operational overhead of Hadoop by removing the single points of failure in the Hadoop NameNode. Second, to offer easy Hadoop integration for Cassandra users (one distributed system is enough).

In order to support massive files in Cassandra we came up with a novel approach that relies heavily on the fundamentals of Cassandra’s architecture, both Dynamo and BigTable.

CFS Data Model

CFS is modeled as a Keyspace with two Column Families in Cassandra.  The Keyspace is where replication settings are, so unlike HDFS, you can’t change replication per file.  You can, however, accomplish the same with multiple CFS Keyspaces. The two Column Families represent the two primary HDFS services. The HDFS NameNode service, that tracks each files metadata and block locations, is replaced with the “inode” column family.  The HDFS DataNode service, that stores file blocks, is replaced with the “sblocks” Column Family. By doing this we have removed three services of the traditional Hadoop stack and replaced them with one fault tolerant scalable component.

The ‘inode’ Column Family contains meta information about a file and uses a DynamicCompositeType comparator.  Meta information includes: filename, parent path, user, group, permissions, filetype and a list of block ids that make up the file. For block ids it uses TimeUUID so blocks are ordered sequentially naturally.  This makes supporting HDFS append() simple.

Secondary indexes are used to support operations like “ls” and “rmdir” the corresponding CQL looks something like:

select filename from inode where parent_path=‘/tmp’;

select filename from inode where filename > ‘/tmp’ and filename < ‘/tmq’ and sentinel = ‘x’;

The sentinel is needed because of how cassandra implements secondary indexes. It chooses the predicate with the most selectivity and filters from there.

Since file identifiers are accessed by secondary indexes it can use a UUID as the row key, giving us good distribution of data across the cluster. To summarize, an entire files meta information is stored under one wide row key.

The second column family ‘sblocks’ stores the actual contents of the file.

Each row represents a block of data associated with an inode record.  The row key is a block TimeUUID from a inode record. The columns are time ordered compressed sub-blocks that, when decompressed and combined, equal one HDFS block.  Let’s walk through the read and write paths to make this all more clear.

CFS Write Path

Hadoop has the “dfs.block.size” parameter to tell how big a file block should be per file write. When a file comes in it writes the static attributes to the inode column family.

Then allocates a new block object and reads dfs.block.size worth of data.   As that data is read it splits it into sub blocks of size “cfs.local.subblock.size”.  those sub-blocks are compressed using google snappy compression.  Once a block is finished the block id is written to the inode row and the sub blocks are written to cassandra with the block id as the row key and the sub block ids as the columns.  CFS splits a block into sub-blocks since it relies on thrift, which does not support streaming so it must ensure it doesn’t OOM the node by sending 256MB or 512Mb of data at once. To Hadoop though, the block looks like a single block so it doesn’t cause any change in the job split logic of map reduce.

CFS Read Path

When a read comes in for a file or part of a file (let’s assume Hadoop looked up the the uuid from the secondary index) it reads the inode info and finds the block and subblock to read.  CFS then executes a custom thrift call that returns either the specified sub-block data or, if the call was made on a node with the data locally, the file and offset information of the Cassandra SSTable file with the subblock.  It does this since during a mapreduce task the jobtracker tries to put each computation on the node with the actual data. By using the SSTable information it is much faster, since the mapper can access the data directly without needing to serialize/deserialize via thrift.

One question that comes up is why does CFS compress sub-blocks and not use the ColumnFamily compression in Cassandra 1.0?  The reason is by compressing and de-compressing on the client side it cuts down of the network traffic between nodes.

Integrating with Hadoop

Hadoop makes it very simple to hook a custom file system implementation in and everything just works. Only the following change is needed to core-site.xml

<property>
 <name>fs.cfs.impl</name>
 <value>com.datastax.bdp.hadoop.cfs.CassandraFileSystem</value>
</property>

Now it’s possible to execute the following commands:

hadoop fs -copyFromLocal /tmp/giant_log.gz cfs://cassandrahost/tmp

or even

hadoop fs distcp hdfs:/// cfs:///

We’re here to help. We’ve set up “Meet the Experts” sessions around the country on regular schedules. Stop by and meet Cassandra developers and experts to get your questions answered.

We’re adding more and more cities to the list, but if one of these doesn’t work for you feel free to stop by our support forums for the digital version.  We’ll answer your questions online with the same level of expertise.