Papers
Cassandra: A Decentralized Structured Storage System
Avinash Lakshman, Prashant Malik · ACM SIGOPS Operating Systems Review 44(2) (versao LADIS 2009 lida), 2010
The Problem: Failure as the Normal Case at Facebook Scale
The Cassandra paper comes from Facebook, which at the time ran a social platform serving hundreds of millions of users at peak times from tens of thousands of servers in many data centers. The authors state that dealing with failures across thousands of components is the standard mode of operation, so software has to treat failure as the norm instead of the exception. Cassandra was built for a concrete workload, Inbox Search, which lets users search their Facebook inbox. That feature demanded very high write throughput, on the order of billions of writes per day, and had to keep scaling as the number of users grew. Because users are served from geographically distributed data centers, replicating data across data centers was also key to keeping search latency low. The paper reports that Inbox Search launched in June 2008 for around 100 million users and, at the time of writing, served over 250 million. The abstract describes the goal as a distributed store for very large amounts of structured data on many commodity servers, with highly available service and no single point of failure. It resembles a database in many design strategies but does not support a full relational data model; it offers a simpler model with dynamic control over data layout and format. The authors describe Cassandra as a synthesis of well-known techniques to achieve scalability and availability.
Data Model and API
A table in Cassandra is described as a distributed multi-dimensional map indexed by a key, whose value is a highly structured object. The row key is a string with no size restriction, typically 16 to 36 bytes long. Every operation under a single row key is atomic per replica, no matter how many columns are read or written. Columns are grouped into column families, an idea the authors say is very similar to Bigtable. There are two kinds of column families: simple, and super, which can be pictured as a column family inside a column family. Applications choose how columns are sorted inside a family, either by time or by name; time sorting suits Inbox Search, where results appear in time order. A simple column is addressed as column family then column, and a column inside a super column family is addressed as column family, super column, then column. The paper notes that applications typically use a dedicated Cassandra cluster of their own, and that although the system supports several tables, every deployment had only one table in its schema. The client API is deliberately tiny, with three methods: insert(table, key, rowMutation), get(table, key, columnName) and delete(table, key, columnName). The column name argument can point to a specific column, a column family, a super column family, or a column within a super column.
Partitioning and Replication
Cassandra partitions data with consistent hashing, but with an order-preserving hash function. The output range is treated as a ring; each node is assigned a random position on it, and a key is hashed to a position and then walked clockwise to the first node with a larger position, which becomes the coordinator for that key. A node therefore owns the region between itself and its predecessor, and the main advantage the paper cites is that a node arriving or leaving only affects its immediate neighbors. The paper names two weaknesses of the basic scheme: random positions give non-uniform data and load distribution, and the algorithm ignores differences in node performance. Dynamo-style multiple ring positions per node is one remedy; Cassandra instead lets lightly loaded nodes move on the ring to relieve heavily loaded ones, because the authors found this made the design tractable and load-balancing choices deterministic. For replication, each item is stored on N hosts, where N is a per-instance replication factor. The coordinator stores the key locally and replicates it to N-1 other nodes. Policies include Rack Unaware, Rack Aware and Datacenter Aware; under Rack Unaware the other replicas are the N-1 successors of the coordinator on the ring. The paper says rows are replicated across multiple data centers so that an entire data center failure can be handled without outage. For requests, in synchronous mode writes wait for a quorum of replicas to acknowledge (the paper notes some high-throughput systems use asynchronous replication instead); reads go either to the closest replica or to all replicas, waiting for a quorum, depending on the consistency the client needs.
Membership and Failure Detection
Cluster membership is based on Scuttlebutt, which the paper describes as a very efficient anti-entropy gossip mechanism, valued for its low CPU use and efficient use of the gossip channel. Gossip also carries other system control state, not only membership. For failure detection the authors use a modified version of the Phi Accrual Failure Detector. Instead of giving a yes-or-no answer, the detector produces a suspicion level, Phi, on a scale that adjusts dynamically to network and load conditions. If a node is suspected when Phi reaches 1, the chance of a mistake, meaning a late heartbeat later contradicts the decision, is about 10 percent; it is about 1 percent at Phi equal to 2 and about 0.1 percent at 3. Each node keeps a sliding window of inter-arrival times of gossip messages from the others and computes Phi from the resulting distribution. The original accrual paper suggests a Gaussian approximation, but the Cassandra authors found an exponential distribution fit better because of how the gossip channel affects latency. They also claim their accrual detector in a gossip setting is, to their knowledge, the first of its kind. In their experience section, other detectors took too long as clusters grew: in one experiment with 100 nodes, detecting a failed node took about two minutes, while the accrual detector with Phi set to 5 brought the average to about 15 seconds.
Local Persistence and the Write Path
On each machine Cassandra relies on the local file system. A write first goes to a commit log for durability and recoverability, and only after that succeeds is an in-memory data structure updated. The authors dedicate a disk to the commit log because its writes are sequential, which maximizes throughput. When the in-memory structure crosses a threshold based on data size and number of objects, it is dumped to disk as a file written sequentially, together with an index by row key. Over time many such files accumulate, and a background merge process, which the paper says is very similar to Bigtable compaction, combines them. The paper says compaction picks files of similar size, and that a periodic major compaction merges all related files into one, an I/O-heavy operation. A read checks the in-memory structure first, then disk files from newest to oldest. To avoid reading files that lack the key, each data file has a bloom filter summarizing its keys, kept in memory too. Column indices are generated at every 256K chunk boundary, a configurable value that worked well for them. Because dumped files are never mutated, reads need no locks, and the authors call the server practically lockless for reads and writes, avoiding the concurrency issues of B-tree based databases. They also describe a fast sync mode in which commit log writes are buffered, which risks data loss if the machine crashes.
Operational Lessons and the Inbox Search Deployment
The paper's experience section opens with the lesson of not adding any feature without understanding how applications will use it, noting that most problematic scenarios do not come from node crashes or partitions alone. Before launching Inbox Search, the team indexed 7TB of inbox data for over 100 million users from MySQL, using Map/Reduce jobs that acted as clients of Cassandra; background channels let those jobs aggregate each user's reverse index so the instance was bottlenecked only by network bandwidth. Most applications needed atomicity only per key per replica, but some asked for transactions, mainly to maintain secondary indices, and the authors said they were working on exposing such atomic operations. They also observed that even a fully decentralized system benefits from some coordination, and they used Zookeeper for tasks such as electing a leader that assigns replica ranges. For Inbox Search the schema had two column families: one keyed by user id for term search, where words are super columns and message identifiers are columns, and one for interactions, where recipient ids are super columns. When a user clicks the search bar, an asynchronous message primes the cache with that user's index. The paper reports about 50+TB on a 150-node cluster across east and west coast data centers, with measured latencies in milliseconds: for interactions, min 7.69, median 15.69, max 26.13; for term search, min 7.78, median 18.27, max 44.41. Planned future work included compression, atomicity across keys and secondary indexes.
See the ideas at work
Practice the designs that use this paper in the System Design school.