Scaling a distributed system

The properties being bought

Algorithms in a distributed system have to work equally well with any number of nodes, because the number changes while the system is running. Two properties are what the whole exercise is for:

• Fault tolerance. Any number of nodes can fail and the cluster keeps functioning as long as one is alive. This is what lets a business run without an interruption every time hardware does.
• Horizontal scalability. Nodes can be added dynamically. This is what lets capacity follow demand rather than being provisioned for the worst case a year in advance.

The algorithms also have to avoid bottlenecks of their own. The herd effect — every node reacting to the same event at the same instant — is the classic one, and it turns a single failure into a system-wide load spike.

Some vocabulary used throughout:

• Node — a process running on a dedicated machine.
• Cluster — a collection of nodes connected to each other, working on the same task and typically running the same code.
• Master node — the node that distributes work among the others and collects the results.

What has to scale

1. User interfaces and external APIs, which sit on user-facing servers. Load balancing and replication are the two usual techniques.
2. The internal coordination layer — the coordination service such as ZooKeeper, load balancers, the service registry, message brokers.
3. The storage layer, which is where most of the difficulty is.

Storage: files or a database

Storing in a filesystem is the low-level, general-purpose option. It takes data of any format, structure or size, and is normally used for unstructured data: video, audio, text files, memory logs.

A database is a higher-level abstraction over the same problem, and what it adds is knowledge of the structure:

• A query language, caching and performance optimisations that exploit that structure
• Constraints applied to the data
• Transactions — relational databases guarantee ACID transactions: atomic, consistent, isolated, durable

Databases come in two families:

1. Relational.
2. Non-relational (NoSQL), for less structured data — key/value pairs, key/document pairs, graph databases.

NoSQL databases are generally easier to scale than SQL databases, because each record is independent of the others and can therefore be moved to another machine without dragging its relationships along. The cost is on the other side: it becomes harder to analyse data across records and across databases, and work the relational engine would have done in a join moves into the application.

Whichever family is used, the database has to be available, scalable and fault tolerant.

Why a single database instance runs out

• It is a single point of failure.
• It is a performance bottleneck:
  • Parallelism is limited to the number of cores in one machine.
  • The number of connections is limited by what the operating system and network card support.
  • Minimum latency is set by the geographical distance between the instance and the user.
  • Working set size is limited to the memory one machine can hold.

Database sharding

Sharding means splitting the data and placing each chunk on a different machine.

Consistent hashing is the common technique. It maps both the keys of the data and the database nodes that store it into the same hash space, so a key’s owner is found by walking the space rather than by consulting a table.

Its benefits are:

1. Nodes can be added to and removed from the cluster without reallocating every key across the new set of nodes — only the keys in the affected range move.
2. More keys can be allocated to some nodes and fewer to others, using the virtual nodes technique, which is what allows a cluster of unequal machines.
3. Keys can be spread more evenly across nodes by using multiple hash functions.

Database replication

Replication means creating identical copies of the data and placing each copy on a different machine. That redundancy buys high availability and fault tolerance, and for read-intensive workloads it buys throughput, because reads can be served by any copy.

Sharding and replication are used together: sharding decides which node owns a key, replication decides how many copies of it exist.

Eventual consistency

In a master–master configuration, eventual consistency is the model: if no further updates are made, every reader eventually sees the newest data, but some readers may temporarily see stale data — for example when a write goes to one node and a read immediately follows on another. It suits systems that do not need the most up-to-date data everywhere at once, such as posting to a social media profile.

The advantage is lower latency and higher availability, because a write does not have to wait for every replica and a read does not have to find the newest one.

A distributed database is eventually consistent when

R + W ≤ N

where:

• R is the number of nodes read from
• W is the number of nodes written to
• N is the total number of replicas

The read set and the write set are not guaranteed to overlap, which is exactly why a read can miss a recent write.

Strict consistency

In a symmetric master–master architecture, strict consistency means waiting for every node to be updated before acknowledging a write. The write is therefore no faster than the slowest replica, and unavailable if any replica is unreachable.

Quorum consensus

A distributed database guarantees strict consistency when

R + W > N

Because the read set and the write set must then share at least one node, every read reaches a node that saw the latest write.

Which of the overlapping copies is the latest is settled by versioning the record. Each record carries a key, its data and a version number, and every update increments the version: a record written as (Key1, Data1, v1) becomes (Key1, Data2, v2). A reader that receives several copies takes the highest version.

Choosing R and W tunes the system for reads or for writes. A low R and a high W makes reads cheap and writes expensive; the reverse makes writes cheap and reads expensive. Both remain strictly consistent as long as R + W > N.