Partitioning, scaling and caching

Replication copies the same data to several machines. Partitioning, also called sharding, splits different data across machines, so that each holds and serves only part of the whole. Replication makes a system survive failures and serve more reads; partitioning makes it hold more data and handle more writes than any single machine could. Large systems use both: data is partitioned, and each partition is replicated, as with the per-range Raft groups of lesson 9. This lesson covers how to divide data among partitions, how to rebalance when machines join or leave, how to route requests to the right partition, how to cope with skewed load, and how caching reduces load on the partitions altogether. CMPUT-404 (lesson 11) covered scaling web front ends; here we focus on the stateful back end, where the hard problems are.

Goals of partitioning

A good partitioning scheme spreads both data and load evenly, so that no partition is a bottleneck; it keeps related data together, so that common queries and transactions touch one partition; it lets the system add or remove machines while moving as little data as possible; and it makes it cheap to find which partition holds a key. These goals conflict. Spreading data randomly balances load but scatters related data; keeping related data together risks hot spots when one group of data is popular.

Range partitioning

In range partitioning, keys are sorted and each partition holds a contiguous range, like volumes of an encyclopedia. Bigtable, HBase, Spanner, CockroachDB and TiKV partition this way. Range scans are efficient, since keys near each other are on the same partition, and partitions can split automatically when they grow too large or too busy. The danger is skew: if keys are timestamps, every new write goes to the partition holding the latest range, making it a hot spot while others idle. Designers avoid this by prefixing keys, for example with a business identifier before the timestamp, so writes for different businesses go to different ranges.

Hash partitioning

In hash partitioning, a hash of the key decides the partition. A good hash function spreads keys uniformly regardless of their patterns, so load is balanced, but range scans become expensive because adjacent keys are scattered. The naive scheme assigns key \(k\) to partition \(\text{hash}(k) \bmod N\), where \(N\) is the number of nodes. Its fatal flaw appears when \(N\) changes: adding one node changes the result of the modulus for almost every key, so nearly all data must move.

Consistent hashing

Consistent hashing, introduced by Karger and colleagues in 1997 for web caching and later used in Dynamo, Cassandra and many content delivery networks, solves the rebalancing problem. Imagine the output space of the hash function as a ring. Each node is hashed to a point on the ring, and each key is assigned to the first node clockwise from the key's own hash. When a node joins, it takes over only the keys between itself and its predecessor; when a node leaves, only its keys move, to its successor. On average only \(1/N\) of the keys move, the minimum possible.

With one point per node, the arcs between nodes vary widely in length, so some nodes get several times the average load. The remedy is virtual nodes: each physical node is hashed to many points, for example 100 or 256, so its share of the ring is the sum of many small arcs and averages out. Virtual nodes also let a more powerful machine take more points, and when a node fails its load spreads over many others rather than landing on a single successor.

# Placing 100,000 keys: hash mod N versus consistent hashing with virtual nodes.
import bisect, hashlib, statistics

def h(s): return int(hashlib.md5(s.encode()).hexdigest()[:12], 16)
KEYS = [f"acct-{i}" for i in range(100_000)]

def mod_n(nodes):
    return {k: nodes[h(k) % len(nodes)] for k in KEYS}

def ring(nodes, vnodes):
    points = sorted((h(f"{n}#{v}"), n) for n in nodes for v in range(vnodes))
    hashes = [p for p, _ in points]
    def owner(k):
        i = bisect.bisect(hashes, h(k)) % len(points)   # first point clockwise from the key
        return points[i][1]
    return {k: owner(k) for k in KEYS}

def moved(a, b): return sum(a[k] != b[k] for k in KEYS) / len(KEYS)
def spread(assign, nodes):
    counts = [sum(1 for v in assign.values() if v == n) for n in nodes]
    return max(counts) / statistics.mean(counts)

before, after = [f"n{i}" for i in range(10)], [f"n{i}" for i in range(11)]
print(f"hash mod N:          adding an 11th node moves {moved(mod_n(before), mod_n(after)):.1%} of keys")
for v in (1, 10, 100):
    a, b = ring(before, v), ring(after, v)
    print(f"ring, {v:3d} vnodes/node: moves {moved(a, b):5.1%} of keys; busiest node holds "
          f"{spread(a, before):.2f}x the average")

hash mod N:          adding an 11th node moves 90.9% of keys
ring,   1 vnodes/node: moves 12.2% of keys; busiest node holds 2.25x the average
ring,  10 vnodes/node: moves  7.0% of keys; busiest node holds 1.43x the average
ring, 100 vnodes/node: moves  8.2% of keys; busiest node holds 1.20x the average

Two related techniques are worth knowing. Rendezvous hashing, or highest random weight hashing (Thaler and Ravishankar, 1996), computes a score for each pair of key and node and assigns the key to the node with the highest score; it also moves only \(1/N\) of keys and needs no ring. And jump consistent hash (Lamping and Veach, 2014) maps keys to numbered buckets with almost no memory, when nodes are only ever added at the end. Many systems, however, abandon hashing to individual nodes altogether in favor of a fixed number of partitions, described next.

Fixed partitions and rebalancing

An alternative used by Elasticsearch, Riak, Couchbase and Kafka is to create many more partitions than nodes up front, say 1,000 partitions for 10 nodes, and assign whole partitions to nodes. Keys map to partitions by hash and never change partition; when a node is added, it takes over some whole partitions from the others, and only those move. This makes rebalancing a matter of copying partitions, which is easy to throttle and reason about. The number of partitions must be chosen with growth in mind: too few and partitions become huge; too many and the per-partition overhead adds up. Range-partitioned systems instead split and merge partitions dynamically as they grow and shrink.

Rebalancing itself must be handled carefully. Moving data consumes network and disk bandwidth and can degrade service, and automatic rebalancing triggered by a node that is merely slow can make matters worse by adding load to an already struggling cluster. Many operators prefer that the system propose a rebalancing plan and a person approve it, or that automatic rebalancing be heavily rate-limited.

Request routing

Once data is partitioned, a request must find the node holding its key. There are three common approaches. Clients can contact any node, which forwards the request to the right one if necessary; Cassandra uses this, with nodes learning the ring through a gossip protocol in which each node periodically exchanges state with a few random peers, so information spreads through the cluster in a logarithmic number of rounds. A separate routing tier can sit in front of the nodes and forward requests; MongoDB's mongos routers work this way. Or clients can be partition-aware, holding a copy of the partition map and contacting the right node directly, as Kafka clients do. In all three, someone must hold the authoritative map of which partitions live where, and that map must change consistently during rebalancing. Many systems keep it in a coordination service such as ZooKeeper or etcd, so that its updates are linearizable.

Skew and hot keys

Even perfect hashing cannot balance a workload in which a single key is extremely popular, because all requests for one key go to one partition. Celebrity accounts on social networks, a flash sale on one product, or one large customer of Ledgerline sending a burst of invoices all create hot keys. Remedies include splitting a hot key into several sub-keys, for example appending a random digit to spread writes over ten partitions and then combining them on read; caching popular reads; queueing and batching writes; and isolating very large customers on dedicated partitions. Detecting skew requires monitoring per-partition load, since cluster-wide averages hide it.

Caching

A cache keeps copies of frequently used data in faster storage, usually memory, closer to where it is used. Caching reduces latency and protects the back end: a cache that answers 95 percent of reads reduces the load on the database twentyfold. Real workloads are usually skewed, following roughly a Zipf distribution, in which the \(r\)-th most popular item is requested in proportion to \(1/r\). Under such skew, a cache holding a small fraction of items can answer a large fraction of requests.

# A cache in front of the ledger: hit rate of an LRU cache under a skewed (Zipf) workload.
import random
from collections import OrderedDict
random.seed(450)
ITEMS = 100_000
weights = [1 / (r ** 1.0) for r in range(1, ITEMS + 1)]      # Zipf with exponent 1
requests = random.choices(range(ITEMS), weights=weights, k=200_000)

def hit_rate(capacity):
    cache, hits = OrderedDict(), 0
    for key in requests:
        if key in cache:
            hits += 1; cache.move_to_end(key)
        else:
            cache[key] = True
            if len(cache) > capacity: cache.popitem(last=False)   # evict least recently used
    return hits / len(requests)

for cap in (100, 1_000, 10_000):
    print(f"cache holds {cap / ITEMS:6.1%} of items -> hit rate {hit_rate(cap):.1%}")
top = sum(weights[:100]) / sum(weights)
print(f"the 100 most popular items receive {top:.1%} of requests (a hot-key risk for one shard)")

cache holds   0.1% of items -> hit rate 28.9%
cache holds   1.0% of items -> hit rate 50.2%
cache holds  10.0% of items -> hit rate 72.5%
the 100 most popular items receive 42.9% of requests (a hot-key risk for one shard)

Caches are organized in several ways. With cache-aside, the application reads from the cache and, on a miss, reads from the database and fills the cache; this is the most common pattern, used with Memcached and Redis. With read-through and write-through caches, the cache itself loads from and writes to the database. With write-back caches, writes go to the cache and are flushed later, which is fast but risks losing writes on a crash. Content delivery networks (CMPUT-313, lesson 4) are caches distributed across the world for static content and, increasingly, dynamic content at the edge.

Cache invalidation and consistency

Phil Karlton's quip that there are only two hard things in computer science, cache invalidation and naming things, is a distributed systems joke. When the underlying data changes, cached copies become stale. Setting a time to live (TTL) bounds staleness but serves stale data until it expires. Deleting the cached entry on every write is better but has races: if a reader fetches an old value from the database, a writer updates the database and deletes the cache entry, and the reader then writes its old value into the cache, the stale value persists until the next write. Facebook's 2013 paper on scaling Memcache described leases to prevent exactly this: on a miss, the cache gives the reader a token, and a later fill is accepted only if no delete has occurred since the token was issued.

Caches also create failure modes. If a popular entry expires, many requests miss at once and all hit the database, a cache stampede; the remedies are to let one request refill the entry while others wait or receive the stale value, and to add jitter to expiry times. If the cache tier fails or is restarted, the database suddenly receives the full load, which it may not survive; such cold cache events have caused many outages, so cache warm-up and gradual traffic ramp-up are part of operating a cached system. A cache, in short, is a performance optimization the system must be able to survive losing.

Scaling compute: stateless services and autoscaling

Partitioning addresses state; the stateless parts of a system scale differently. A service that keeps no data between requests can be replicated freely behind a load balancer, and instances can be added or removed in response to load, the basis of autoscaling in cloud platforms and Kubernetes. Load balancers spread requests using round robin, least connections, or power of two choices: picking two instances at random and sending the request to the less loaded one, which, as Mitzenmacher showed in 1996, balances load exponentially better than a single random choice at almost no cost. Sticky sessions, which route a user to the same instance, reintroduce state and should be avoided where possible by keeping session data in a shared store.

Worked examples