Problem RestatementProblem
A huge multiset (a list of values with repeats) is split across W worker machines. Find the mode, the value that appears most often, exactly, without sending every raw item to one machine (too much network and memory). Anthropic asked this.
Deep Dive — Finding the most frequent value across W machinesDeep dive
The data is already spread across workers. The mode is a global property, so some communication is unavoidable — the question is how little.
Send everything to one machine
Ship all the items to a coordinator and count there.
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
W1["Worker 1 - billions of items"] --> CO["Coordinator"]
W2["Worker 2"] --> CO
W3["Worker W"] --> CO
CO --> NET["Every item crosses the network"]
CO --> MEM["One machine must hold the full counter map"]
MEM --> FAIL["Out of memory, and the network is the bottleneck anyway"]Correct and unrunnable. It moves O(total items) bytes and requires one machine to have memory proportional to the number of distinct values — the two things distributing the data was meant to avoid.
Each worker sends its local mode
Every worker counts locally, sends only its most frequent value, and the coordinator takes the winner. Network cost drops to W messages.
It is also wrong, and the counterexample is short enough to say out loud:
worker 1: {a: 10, b: 9} local mode a (10)
worker 2: {c: 10, b: 9} local mode c (10)
globally: b = 18 -> b is the mode, and neither worker reported itA value can be second everywhere and first overall. No amount of top-k per worker fixes this in general — only exact global counts do.
Shuffle by value, so each value is counted in one place
Partition by the value, not by the worker that holds it:
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
W1["Worker 1"] -->|"local counts: (a,10) (b,9)"| H["Partition by hash(value) % R"]
W2["Worker 2"] -->|"(c,10) (b,9)"| H
H --> R1["Reducer 1 - owns b: 9 + 9 = 18"]
H --> R2["Reducer 2 - owns a: 10, c: 10"]
R1 --> M["Each reducer reports its local max"]
R2 --> M
M --> ANS["Global max of R candidates - exact"]Every occurrence of a value lands on the same reducer, so its total is complete and the comparison is valid. Two properties make this practical:
- Combine before shuffling. Workers send
(value, count)pairs, not raw items, so the network carries O(distinct values) rather than O(items) — usually orders of magnitude less. - Memory is per reducer. Each holds only the values hashed to it, so distinct-value capacity scales with the number of reducers.
Exact Algorithm (MapReduce style)
- Local count (combiner): each worker counts its own items with a hash map,
value → count. This shrinks the data a lot when values repeat. - Shuffle by value: send each
(value, local_count)to the workerhash(value) % W. Now all partial counts for a value arrive at the same worker. - Reduce: each worker sums the counts for the values it owns → exact global counts for its share of values → its local max (value, count).
- Final: each worker sends just its best
(value, count)to a coordinator, which picks the overall max. That's only W small messages.
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
W1["Worker 1: count locally"] -->|"hash(value)"| R1["Reducer A"]
W1 --> R2["Reducer B"]
W2["Worker 2: count locally"] --> R1
W2 --> R2
W3["Worker 3: count locally"] --> R1
W3 --> R2
R1 -->|"best (value, count)"| CO["Coordinator - global max"]
R2 --> COThis is correct because after the shuffle, every value's full count lives on exactly one reducer, so the max over reducers' maxima is the true mode.
Cost
- Network = the number of distinct (value, worker) pairs after local counting. When there are few distinct values, it's tiny. When almost all values are unique, it's close to the raw data (but then the mode's count is small anyway).
- Memory: each reducer holds counts only for its hash range. If it doesn't fit, sort and spill to disk (external aggregation), or use more reducers.
Handling Skew and ScaleScale
- Skew: one value might be extremely common, sending huge partial counts to one reducer. Local combining already solves most of this (each worker sends one number per value, not every occurrence).
- Pruning trick (optional): each worker reports its top-k locally with counts, and a threshold algorithm can prove the winner without the full shuffle in many cases (the sum of the remaining possible counts can't beat the current best). It's useful when the data is very skewed.
- Approximate alternative: if an exact answer isn't required, each worker builds a Count-Min Sketch (fixed small memory, mergeable by adding arrays) plus a small heavy-hitter list. Merge the sketches and pick the top candidate. It needs far less network, with a small, bounded error.
Code Sketch (single process simulating workers)
from collections import Counter
def distributed_mode(partitions, num_reducers=4):
reducers = [Counter() for _ in range(num_reducers)]
for part in partitions: # each "worker"
local = Counter(part) # 1) combine locally
for value, c in local.items(): # 2) shuffle by hash
reducers[hash(value) % num_reducers][value] += c
best = [r.most_common(1)[0] for r in reducers if r] # 3) reducer maxima
return max(best, key=lambda vc: vc[1]) # 4) global max
print(distributed_mode([["a"]*10 + ["b"]*9, ["c"]*10 + ["b"]*9])) # ('b', 18)Wrap-UpWrap-up
Local modes can be wrong, so count locally, hash-partition the (value, count) pairs so each value's complete count lands on one reducer, sum them there, take each reducer's max, and pick the global max from those few candidates. Local combining keeps network cost low and tames skew, spilling handles memory limits, and sketches give a cheap approximate answer when exactness isn't needed.