Skip to main content

Scatter/Gather

Adapted from: Brendan Burns, Designing Distributed Systems (O'Reilly), Chapter 8 — rewritten and condensed for teaching.

Replicated load-balanced services scale request rate. Sharded services scale data size. Scatter/gather scales time: it farms one request out to many workers in parallel, then merges partial answers so wall-clock latency drops versus doing the work sequentially.

Like those patterns, scatter/gather is a tree: a root distributes work; leaves process slices. Unlike ordinary sharding — where one shard handles a request — scatter/gather typically sends work to many (or all) leaves at once. Each leaf returns a fraction of the answer; the root combines them and replies to the client.

flowchart TB
  CLIENT[Client]
  ROOT[Root]
  L1[Leaf 1]
  L2[Leaf 2]
  L3[Leaf 3]
  CLIENT -->|request| ROOT
  ROOT -->|scatter| L1
  ROOT -->|scatter| L2
  ROOT -->|scatter| L3
  L1 -->|partial| ROOT
  L2 -->|partial| ROOT
  L3 -->|partial| ROOT
  ROOT -->|complete answer| CLIENT

Use scatter/gather when a request needs a large amount of mostly independent computation (or index lookup) that you can split, run in parallel, and stitch back together.

Scatter/Gather with Root Distribution​

The simplest form: every leaf can see the same data (or the same full index), but the root splits the computation across leaves — an “embarrassingly parallel” problem.

Example: one core needs 60 s for request R. On a 30-core machine, ideal parallelization is about 2 s — still slow for a web request, and a single box hits memory, disk, and NIC limits. Scatter/gather pushes the same idea across machines: more cores, more bandwidth pools, and a root that can favor responsive leaves when one node is slow (noisy neighbor).

Hands On: Distributed Document Search (term scatter)​

Goal: find documents containing both "cat" and "dog".

A full-text index is a map from word → list of document IDs. Looking up "cat" and "dog" are independent. The root only needs both lists to compute the intersection.

flowchart LR
  ROOT[Root]
  CAT[Leaf: cat]
  DOG[Leaf: dog]
  ROOT -->|cat| CAT
  ROOT -->|dog| DOG
  CAT -->|doc1,doc2,doc4| ROOT
  DOG -->|doc1,doc3,doc4| ROOT
  ROOT -->|intersect → doc1,doc4| OUT[Client]

Implement the root merge for term-sharded scatter/gather: given one document-ID list per word, return the sorted intersection.

Check: root distribution​

Quiz

Root distribution

1 / 2

In term-sharded document search for "cat" AND "dog", why can the two lookups run on different leaves?

Pick an answer to see if you are right.

Scatter/Gather with Leaf Sharding​

Root distribution alone does not grow past what one machine can store. When the corpus is huge (e.g. all patents worldwide), shard the documents across leaves.

Unlike request-level sharding — where the router sends each request to one shard — leaf-sharded scatter/gather sends the query to every shard. Each leaf searches its slice and returns matches. The root unions (or otherwise merges) those partial hits into one response.

Prefer a stable shard function (e.g. patent_id % num_shards) over contiguous ID ranges that always dump new data onto the newest shard.

Each leaf holds only some documents. For "cat" AND "dog", every leaf returns docs in its shard that contain both words. The root builds the union of those sets (not the intersection of different words — each leaf already applied the AND locally).

flowchart TB
  ROOT[Root]
  S0["Shard 0: docs 1–10 → doc1, doc5"]
  S1["Shard 1: docs 11–20 → doc15"]
  S2["Shard 2: docs 21–30 → doc22, doc28"]
  ROOT --> S0
  ROOT --> S1
  ROOT --> S2
  S0 --> ROOT
  S1 --> ROOT
  S2 --> ROOT
  ROOT -->|union| OUT["doc1, doc5, doc15, doc22, doc28"]

Implement shard-local AND search and root-side union.

Check: leaf sharding vs term scatter​

Quiz

Leaf sharding

1 / 2

In leaf-sharded scatter/gather for a conjunctive query, what does the root typically merge?

Pick an answer to see if you are right.

Choosing the Right Number of Leaves​

More leaves are not always faster.

Constant overhead vs useful work​

Every leaf pays fixed costs (parse request, network, TLS, …). Extra leaves raise total overhead while shrinking useful work per leaf. Past a point, overhead dominates — parallel speedup is asymptotic, not unlimited.

The straggler problem​

The root usually waits for all leaf responses. Latency of the user request ≈ latency of the slowest leaf.

If a single leaf call has a 1% chance of taking 2 seconds (99th percentile = 2 s), then:

  • 5 parallel leaf calls → about (1 - 0.99^5 ≈ 5%) chance that at least one is a 2 s straggler — your former p99 becomes roughly p95 for the whole fan-out.
  • 100 leaves → you almost always hit a 2 s leaf — overall latency sticks near 2 s.

Availability has the same math: if each leaf fails with probability 1/100 and you need all 100, you are nearly guaranteed a failed user request unless leaves are replicated.

Check: leaves and stragglers​

Quiz

Leaf count and stragglers

1 / 3

Why can adding leaf nodes stop helping (or even hurt) scatter/gather latency?

Pick an answer to see if you are right.

Scaling Scatter/Gather for Reliability and Scale​

A single replica per shard is fragile: one leaf down → every scatter/gather request that needs that shard fails; upgrades under load become hard; RPS is capped by one machine per shard. You also cannot fix compute capacity merely by adding infinite shards — stragglers and overhead intervene.

Replicated sharded scatter/gather: each leaf shard is itself a small replicated service. The root load-balances each shard request across healthy replicas of that shard.

flowchart TB
  ROOT[Root]
  subgraph shard0 [Shard 0 replicas]
    A1[Replica]
    A2[Replica]
  end
  subgraph shard1 [Shard 1 replicas]
    B1[Replica]
    B2[Replica]
  end
  subgraph shard2 [Shard 2 replicas]
    C1[Replica]
    C2[Replica]
  end
  ROOT --> shard0
  ROOT --> shard1
  ROOT --> shard2

Benefits:

  • Leaf failure ≠ whole-query outage (another replica of that shard serves).
  • Rolling upgrades one replica at a time (even across shards in parallel if you choose).
  • Scale hot shards by adding replicas where load concentrates.

Check: replicated shards​

Quiz

Replicated scatter/gather leaves

1 / 1

Why replicate each scatter/gather leaf shard?

Pick an answer to see if you are right.

Key takeaways
  • Scatter/gather parallelizes one request across many leaves, then merges partial answers — scaling for latency, not only RPS or dataset size.
  • Root distribution splits independent work (e.g. per-term lookups) when leaves share a full index; the root intersects or otherwise combines partials.
  • Leaf sharding puts different data on each leaf; the query hits all shards; the root usually unions shard-local matches.
  • More leaves ≠ always faster — fixed per-leaf overhead and stragglers (root waits on the slowest) bound parallelism; p99 leaf latency dominates end-to-end SLOs.
  • Replicate each shard so fan-out stays available under failure and upgrades, and so you can scale compute per partition.
  • Large production systems often compose patterns: replication + sharding + scatter/gather together.

Further Reading​