Cassandra - A Decentralized Structured Storage System

In one breath — the whole paper, compressed

Facebook's Inbox Search needed a store that could take billions of writes per day, stay available while machines died continuously, and span east and west coast data centers. Cassandra answers by joining two lineages: data is partitioned over a consistent-hashing ring and replicated at N nodes with quorum reads and writes, Dynamo-style, while the records themselves are a multi-dimensional map of column families and super columns, Bigtable-style. Every write is a sequential append to a commit log followed by an update to an in-memory structure that is later flushed as one immutable, index-bearing file, with a background compaction merging those files; nothing on disk is ever mutated, so the server is practically lockless for reads and writes. Membership and control state ride on Scuttlebutt anti-entropy gossip, and liveness is decided by a Phi accrual failure detector that emits a continuous suspicion level rather than an up/down bit. In production the Inbox Search cluster held over 50TB on 150 nodes, with median read latencies of 15.69 ms and 18.27 ms.

Before this paper — the world it landed in

By 2008 Facebook served hundreds of millions of users from tens of thousands of servers, and Inbox Search had to be built on top of 7TB of message data that already lived in MySQL. Replicated relational systems gave strong consistency but, as Gray and Helland had argued, at the cost of scalability and availability, and they simply could not keep serving through a network partition. Amazon's Dynamo had shown that a gossiping consistent-hashing ring with client-visible conflict resolution could stay available, but its vector-clock scheme required a read to accompany every write, which is punishing when writes vastly outnumber reads. Google's Bigtable had shown how to give applications a sparse, sorted, column-family data model, but it leaned on GFS for durability and a Chubby-backed master for coordination. Cassandra was written into that gap: Dynamo's availability story with Bigtable's data model, and no distributed file system underneath.

The problem — what was actually breaking

Inbox Search required the system to absorb a very high write throughput, billions of writes per day, and to keep scaling as the user base grew from around 100 million at launch toward over 250 million.
In an infrastructure of thousands of components there is always a small but significant number of servers and network links failing, so the storage system had to treat failure as the norm rather than the exception and expose no single point of failure.
Users are served from geographically distributed data centers, so rows had to be replicated across data centers both to keep search latency down and to survive losing an entire data center to power, cooling, network, or natural-disaster events.
Traditional replicated relational databases focus on guaranteeing strong consistency, which the authors note limits their scalability and availability and leaves them unable to operate through network partitions.
Dynamo's vector-clock conflict detection forces a read to be performed as part of every write, which the authors call very limiting in an environment that must sustain very high write throughput.
Off-the-shelf gossip-style failure detectors degraded with cluster size: in one 100-node experiment the time to detect a failed node was on the order of two minutes, which the authors call practically unworkable in their environment.

Core ideas — the contributions, and why they work

Dynamo distribution, Bigtable data model

Cassandra is explicitly a synthesis of well known techniques rather than a new algorithm: the distribution layer is Dynamo's - a consistent-hashing ring, replication factor N, preference lists, quorum reads and writes, gossip membership - while the record layer is Bigtable's, a distributed multi-dimensional map of column families, with Super column families adding a second level of nesting. The two layers solve orthogonal problems, which is why the graft works: the ring decides which machines hold a key and how the system behaves when they die, and the column family decides how an application lays out and sorts data under one key. Applications get atomicity per row key per replica no matter how many columns are read or written, plus a choice of sorting columns by name or by time, which Inbox Search uses directly so that results come back already in time order. Unlike Bigtable there is no distributed file system underneath: durability is the local file system on every node.

Order-preserving ring, rebalanced by load

The output range of the hash function is treated as a fixed circular space; each node takes a random token as its position, and a key belongs to the first node clockwise from the key's own position, which becomes that key's coordinator and owns the ring segment between itself and its predecessor. Cassandra deliberately uses an order-preserving hash function, so key order survives partitioning, while keeping consistent hashing's central virtue that a node arriving or departing perturbs only its immediate neighbours. Random tokens produce non-uniform load and ignore hardware heterogeneity, and here Cassandra diverges from Dynamo: rather than giving each node many virtual positions, it analyses load information on the ring and moves lightly loaded nodes to relieve heavily loaded ones, the approach described for Chord. The authors pick this because it keeps the design and implementation tractable and makes load-balancing choices deterministic.

A write path that is sequential only

A write is a sequential append to a commit log on a dedicated disk, and only after that append succeeds is the in-memory structure updated; when that structure crosses a threshold computed from data size and object count, it is dumped to a commodity disk in one sequential pass together with a row-key index that is persisted alongside the data. Those files are never mutated afterwards, so a background merge process - compaction, taken from Bigtable - is what consolidates versions and reclaims space. This is fast because every disk operation on the write path is sequential, which is what commodity disks are good at, and because immutability means readers never contend with writers. The authors state that the server instance is practically lockless for read and write operations, which is exactly why they escape the concurrency problems of B-tree based database implementations.

Quorums without read-before-write

A read or write for a key can be sent to any node in the cluster; that node determines the replicas, routes writes to all of them and waits for a quorum of acknowledgements, while a read is either sent to the closest replica or fanned out to all replicas with a quorum wait, depending on the consistency the client asks for. Reconciliation is by timestamp: the routing state machine picks the latest response and schedules a repair on any replica that is behind, so no read is needed to construct a version stamp before writing, which is precisely the Dynamo cost the authors wanted to avoid. Writes can be configured synchronous or asynchronous, and deployments where writes far exceed reads use asynchronous replication. Durability under node failure and network partition comes from relaxing the quorum requirement rather than from blocking until every replica is present.

Accrual failure detection over gossip

Membership and other control state spread through Scuttlebutt, an anti-entropy gossip protocol chosen for efficient CPU usage and efficient use of the gossip channel. On top of it each node keeps a sliding window of the inter-arrival times of gossip messages from every other node, fits a distribution to that window, and emits a continuous suspicion level Phi rather than a boolean verdict, where suspecting at Phi = 1 gives about a 10 percent chance of being wrong, Phi = 2 about 1 percent, and Phi = 3 about 0.1 percent. The value of this is that the threshold becomes an explicit accuracy-versus-speed dial that adapts itself to network and server load, instead of a fixed timeout that must be retuned as the cluster grows. The authors modify the original detector by approximating the inter-arrival distribution as Exponential rather than Gaussian, because that matches the behaviour of the gossip channel, and believe theirs is the first accrual detector deployed in a gossip-based setting.

Topology-aware replication with an elected leader

Each item is replicated at N hosts, N being configured per instance; the coordinator stores the key in its own range and replicates it at N-1 further nodes chosen by an application-selected policy. Rack Unaware simply takes the N-1 successors on the ring, while Rack Aware and Datacenter Aware place replicas deliberately across racks and across data centers, which means placement is no longer derivable from the ring alone. Cassandra therefore elects a leader using Zookeeper; every joining node contacts the leader to learn which ranges it replicates, and the leader maintains the invariant that no node is responsible for more than N-1 ranges. Range metadata is cached on each node and stored fault-tolerantly in Zookeeper so a crashed node comes back knowing its responsibilities, and building the preference list across data centers connected by high-speed links is what lets Facebook lose an entire data center without an outage.

How it works — the mechanism, concretely

Request routing state machine

Any node in the cluster can be the entry point for a request. Its routing state machine walks five states: identify the nodes that own the data for the key; route the requests and wait for responses; fail the request back to the client if replies do not arrive within a configured timeout; figure out the latest response by timestamp; and schedule a repair of the data at any replica that does not have the latest version. All system control messages use UDP while replication and request-routing messages use TCP, over a network layer built on non-blocking I/O. The message-processing pipeline and the task pipeline are split into multiple stages along the lines of SEDA, and the partitioning, membership and failure detection, and storage engine modules were all written from the ground up in Java.

Bootstrapping into the ring

On first start a node chooses a random token for its ring position and persists that mapping both to local disk and to Zookeeper, then gossips it; because every node ends up knowing every other node's token, any node can route a key to the correct owner. A node joining an existing cluster reads a configuration file listing seeds, a few initial contact points, which may also come from a configuration service such as Zookeeper. Every message carries the cluster name of the Cassandra instance, so a node misconfigured to join the wrong cluster is rejected. Because outages at Facebook are usually transient and rarely mean a permanent departure, node addition and removal are explicit administrative operations issued through a command line tool or a browser, which stops a transient failure from triggering partition reassignment or repair of unreachable replicas.

Scaling out and data streaming

A new node is assigned a token chosen so that it alleviates a heavily loaded node, which splits the range that node previously owned. The bootstrap is initiated from any other node in the system by an operator using the command line utility or the Cassandra web dashboard. The node giving up the data streams it to the newcomer using kernel-to-kernel copy techniques, and operational experience puts that rate at 40 MB/sec from a single donor node. The authors report working on having multiple replicas take part in the bootstrap transfer so the effort is parallelized, in the spirit of Bittorrent.

Failure detection loop

Each node maintains, for each peer, a sliding window of the inter-arrival times of gossip messages from that peer, determines their distribution, and computes Phi from it. Callers compare Phi to a threshold to decide whether to treat a peer as down, and the verdict is used not just for membership bookkeeping but to avoid attempting communication with unreachable nodes during ordinary operations. Because Phi is a suspicion level rather than a boolean, different subsystems can pick different confidence points on the same underlying signal. Facebook runs a slightly conservative threshold of Phi = 5, at which failures in a 100-node cluster were detected in about 15 seconds on average.

Commit log, in-memory table, flush

Writes append to a commit log on a machine-dedicated disk so that all commit-log I/O is sequential and the disk's throughput is fully exploited, and the in-memory structure is only updated after the log write succeeds. Commit logs are rolled once they exceed a configurable size, which production experience set at 128MB. Each commit log carries a fixed-size bit-vector header with more bits than the system will ever have column families; when a column family's in-memory structure is flushed to disk its bit is set, and when a log rolls, its bit vector and those of all earlier logs are checked so that fully persisted logs can be deleted. A fast sync mode buffers both the commit-log writes and the in-memory dump, which the authors note implies a potential of data loss on machine crash.

On-disk layout and read path

All data is indexed on the primary key, and each data file is broken into blocks of at most 128 keys, each block demarcated by a block index that records the relative offset of a key within the block and the size of its data; that index is written out at flush time and also kept in memory for fast access. A read always looks in the in-memory structure first, since it holds the newest data for a key, and only then does disk I/O against the data files in reverse time order, returning on the first hit. Each data file carries a Bloom filter summarizing its keys, also held in memory, which is consulted first so that files that cannot contain the key are never opened. Because a key in a column family can have very many columns, column indices are generated at every 256K chunk boundary as columns are serialized out, so a read can jump straight to the right chunk instead of scanning every column; the boundary is configurable but 256K worked well in production.

Compaction

Since flushed files accumulate on disk over time, a background process merges several of them into one, essentially a merge sort over sorted data files, very much like Bigtable's compaction. Cassandra only compacts files that are close to each other in size: the paper states there will never be a situation where a 100GB file is compacted with a file smaller than 50GB, which bounds the cost of any individual merge. Periodically a major compaction runs to collapse all related data files into one big file. The authors concede that compaction is a disk-I/O intensive operation and that many optimizations can be put in place so that incoming reads are not affected.

What the paper showed — measurements and proofs

The Facebook Inbox Search deployment stored about 50+TB of data on a 150-node cluster spread between east and west coast data centers.
Production-measured read latencies for Inbox Search: search interactions at 7.69 ms minimum, 15.69 ms median, 26.13 ms maximum; term search at 7.78 ms minimum, 18.27 ms median, 44.41 ms maximum.
With the accrual failure detector at a slightly conservative Phi threshold of 5, average time to detect a failure in a 100-node cluster was about 15 seconds, against roughly two minutes for the earlier gossip-style detectors the team tried in the same experiment.
Bootstrap data transfer between nodes, using kernel-to-kernel copy techniques, was measured at 40 MB/sec from a single donor node.
The initial load indexed 7TB of inbox data for over 100 million users out of Facebook's MySQL infrastructure using Map/Reduce jobs, sending the serialized reverse index over background channels so that the Cassandra instance was bottlenecked only by network bandwidth.
Operational constants reported from production: commit logs rolled at 128MB, column indices generated at every 256K chunk boundary, and data-file blocks of at most 128 keys; Inbox Search launched in June 2008 for around 100 million users and served over 250 million by the time of writing.

Limits and trade-offs — conceded and discovered

Conceded by the paper: atomicity is only per key per replica, with no transactions across keys. Applications asked for transactional support mainly to maintain secondary indices, and the authors say only that they are working on a mechanism to expose such atomic operations.
Conceded by the paper: compression, atomicity across keys and secondary index support are all listed as future work in the conclusion, so an application that wants an inverted index must build and maintain it itself, as Inbox Search does with super columns.
Conceded by the paper: the fast sync commit-log mode buffers both the log writes and the in-memory dump, which the authors state implies a potential of data loss on machine crash - durability is a tunable, not a guarantee.
Conceded by the paper: although Cassandra is described as a completely decentralized system, the authors learned that some coordination is essential, so Zookeeper is used for leader election and range assignment, and membership changes remain explicit administrative commands rather than automatic reactions.
Exposed later: the order-preserving partitioner with operator-driven token movement produced hot spots and painful rebalancing, and Cassandra eventually defaulted to a random partitioner and adopted Dynamo-style virtual nodes. Reconciling replicas purely by timestamp also means last writer wins, so concurrent updates can be lost under clock skew, a trade-off the paper never names; and the evaluation itself is thin, reporting read latency for one application with no throughput or scalability curves.

What it became — the systems that inherited it

Facebook open-sourced Cassandra in 2008; it entered the Apache Incubator in 2009 and became a top-level Apache project in 2010, which is how an internal Facebook store became the reference implementation of the Dynamo-plus-Bigtable design. The template in this paper - a token ring, replication factor N with per-request tunable consistency, gossip membership, read repair (step five of the routing state machine), and an LSM-structured local engine of commit log, memtable, immutable data files and compaction - is essentially the architecture Apache Cassandra still ships, and it is the canonical case study for leaderless replication in Designing Data-Intensive Applications. Later versions replaced the parts the paper had settled pragmatically: the order-preserving partitioner gave way to a random partitioner, virtual nodes arrived in Cassandra 1.2, Zookeeper was dropped in favour of gossip-based agreement, and the three-method Thrift API of insert, get and delete was superseded by CQL. Its descendants and relatives include ScyllaDB, a shard-per-core C++ rewrite that is wire-compatible with Cassandra, Amazon Keyspaces and DataStax Enterprise as managed services, and Riak, which carried the same Dynamo lineage into the pure key-value world. The Phi accrual failure detector, obscure before this paper, became a standard building block for cluster membership and was adopted by systems such as Akka Cluster. Facebook itself later moved Messages and Inbox Search onto HBase, a reminder that the motivating application outlived its first storage engine while the engine went on to outlive the application.

In the paper’s words — verbatim

“Cassandra system was designed to run on cheap commodity hardware and handle high write throughput while not sacrificing read efficiency.”

Abstract

“Cassandra morphs all writes to disk into sequential writes thus maximizing disk write throughput.”

§5.7

“With the accrual failure detector with a slightly conservative value of PHI, set to 5, the average time to detect failures in the above experiment was about 15 seconds.”

§6

Vocabulary — as this paper uses it

Column family: A named group of columns under a row key, and the unit of organization both in memory and on disk, since Cassandra keeps one in-memory structure and one data file per column family. A column is addressed with the convention column family : column.
Super column family: A column family nested inside a column family, addressed as column family : super column : column. Inbox Search uses message words or recipient ids as super columns and individual message identifiers as the columns inside them.
Coordinator: The node reached by hashing a key onto the ring and walking clockwise to the first node with a larger position. It is responsible for the ring region between itself and its predecessor, and for replicating the keys in that region to the other replicas.
Preference list: Borrowed from Dynamo parlance, the set of nodes responsible for a given range. Cassandra constructs the preference list of a key so that its storage nodes are spread across multiple data centers, allowing a whole data center to fail without an outage.
Phi accrual failure detector: A failure detector that emits a continuously varying suspicion level Phi instead of a boolean up-or-down verdict, computed from a sliding window of gossip inter-arrival times. Raising the threshold trades detection speed for a lower probability of wrongly suspecting a live node.
Scuttlebutt: The anti-entropy gossip mechanism Cassandra uses for cluster membership and for disseminating other system control state, chosen for its efficient CPU utilization and efficient utilization of the gossip channel.
Commit log: The per-node sequential durability log, on its own dedicated disk, that every write must reach before the in-memory structure is updated. Logs roll at a configurable size of 128MB and are deleted once their header bit vector shows every column family they contain has been flushed to disk.
Compaction: The background merge that collates many immutable on-disk data files into fewer, a merge sort over sorted files that only combines files of comparable size, with a periodic major compaction that merges all related files into one.
Bloom filter: A compact summary of the keys in a data file, stored with the file and kept in memory, consulted before a disk lookup so that files which cannot contain the requested key are never read.

On the timeline — where this sits in the story

Year2008–2010

MilestoneCassandra and wide-column NoSQL

Problem solvedLarge web applications needed high write throughput, high availability, and no single point of failure.

How it solved itCassandra combined Dynamo-like distribution with Bigtable-like data modeling.

Why it matteredBecame a major open-source distributed database for write-heavy and always-on workloads.

DDIACh. 4: Storage & Retrieval Ch. 6: Replication Ch. 7: Sharding

View on the timeline