Distributed Sorting and Joins

"I had nine reducers and a perfectly balanced plan. Then one key showed up four million times and decided that reducer three was its permanent home."

A Shuffle That Met a Hot Key

The previous section built aggregation, filtering, and secondary sorting on top of the MapReduce shuffle. This section takes the two heaviest data-preparation operations an AI pipeline runs before any model sees a batch: putting records in a global order, and combining two datasets on a shared key. A feature pipeline almost never trains on raw events. It joins a user table against a clickstream, a product catalog against an impression log, a sensor registry against a telemetry feed, and then sorts the result so that downstream sampling, deduplication, and windowing are cheap. When the datasets are too large for one machine, these become distributed-sort and distributed-join problems, and the cost of doing them badly is measured in cluster-hours of idle workers waiting on one overloaded straggler.

1. Total-Order Distributed Sort Intermediate

Sorting within one reducer is trivial: the framework already delivers each reducer its keys in sorted order. The hard part is a total order across reducers, so that concatenating reducer 0, then reducer 1, then reducer 2 yields one globally sorted stream. A hash partitioner cannot do this; hashing scatters adjacent keys to arbitrary reducers. What we need is a range partitioner: reducer 0 takes the smallest slice of the key space, reducer 1 the next, and so on. The output files, concatenated in reducer order, are then globally sorted with no final merge step.

The catch is choosing the range boundaries. If we split the key space into equal-width intervals, a skewed distribution sends most records to one reducer. The fix, used by the canonical TeraSort benchmark that sorts a terabyte across a cluster, is to sample the data first. Draw a small random sample, sort it, and read off the boundaries at evenly spaced quantiles of the sample. With $R$ reducers we want $R - 1$ split points placed so that each of the $R$ resulting ranges holds roughly $N / R$ records. If the sample has size $m$, the split point for reducer boundary $j$ is the order statistic at rank

Because the sample is drawn from the same distribution as the data, the quantiles of the sample approximate the quantiles of the full dataset, so each reducer receives a near-equal share whatever the underlying distribution looks like. This is the entire idea: a cheap sampling pass buys a balanced range partition, and a balanced range partition turns a global sort into independent local sorts with no merge. We quantify why balance matters so much in the cost models of Chapter 3; the short version is that the slowest reducer sets the wall-clock, so an unbalanced sort runs at the speed of its busiest machine.

2. Two Ways to Join Across Machines Intermediate

A join combines rows from two datasets that share a key: a left dataset $A$ and a right dataset $B$, emitting one output row for every pair $(a, b)$ where $a$ and $b$ carry the same join key. On one machine this is a hash-table lookup. Across machines the difficulty is physical: the matching rows of $A$ and $B$ start on different nodes, and a join can only happen where both rows are present. There are two ways to make them present, and choosing between them is the single most important join decision you make.

The reduce-side join (also called the repartition join) is the general method. The map phase reads both inputs and tags each row with which dataset it came from, emitting the join key as the shuffle key. The shuffle co-locates every row of $A$ and every row of $B$ that share a key onto the same reducer, by the same partitioning that drives every other MapReduce operation. The reducer then separates the two tagged groups and emits their cross product. It works for any two datasets of any size, and it pays for that generality by shuffling both inputs across the network in full.

The map-side join (also called the broadcast or replicated join) is the fast path available when one dataset is small. Instead of shuffling both inputs, you ship the entire small dataset to every mapper, where it becomes an in-memory lookup table. Each mapper then streams its shard of the large dataset and probes the lookup table directly. The large dataset never moves; only the small one is replicated. Sending one dataset to every worker is exactly the broadcast collective introduced in Chapter 4, reused here as a data-processing primitive: the same one-to-all movement that distributes model weights to workers also distributes a lookup table to mappers. Figure 6.5.1 contrasts the two dataflows.

The decision rule is mechanical. If one dataset is small enough to fit in a worker's memory (a few hundred megabytes is a typical threshold), broadcast it and do a map-side join; the large dataset never touches the network. If both datasets are large, you have no choice but the reduce-side join and its full double shuffle. Query engines like Spark, which Chapter 7 takes up, make this choice automatically from table-size statistics, but the underlying mechanics are exactly the two paths in Figure 6.5.1.

3. Both Joins, Side by Side in Code Intermediate

The code below implements both joins over the same two small tables: a tiny users entity table and a larger events fact table, the canonical shape of an AI feature join. The reduce-side path tags both inputs, partitions by a hash of the key, and emits the cross product inside each reducer. The broadcast path turns users into a dictionary shipped to every mapper and probes it while streaming events. We then check that the two strategies produce identical output, which they must: the join result is defined by the data, not by the route the rows took to meet.

import collections, random

users = [("u1", "Ada"), ("u2", "Lin"), ("u3", "Mat")]          # entity table (small)
events = [("u1", "click"), ("u2", "view"), ("u1", "purchase"),
          ("u3", "click"), ("u2", "click"), ("u1", "view")]    # fact table (large)

# ---- Reduce-side join: tag both inputs, shuffle by key, emit cross product ----
def map_tag(rows, side):
    for key, payload in rows:
        yield key, (side, payload)               # tag which dataset the row is from

mapped = list(map_tag(users, "U")) + list(map_tag(events, "E"))

R = 3
buckets = collections.defaultdict(list)
for key, val in mapped:
    buckets[hash(key) % R].append((key, val))    # shuffle: hash partition to R reducers

def reduce_join(pairs):                           # runs inside one reducer
    by_key = collections.defaultdict(lambda: {"U": [], "E": []})
    for key, (side, payload) in pairs:
        by_key[key][side].append(payload)
    for key, sides in by_key.items():
        for u in sides["U"]:
            for e in sides["E"]:
                yield (key, u, e)                 # cross product of the two groups

reduce_side = sorted(row for r in range(R) for row in reduce_join(buckets[r]))

# ---- Broadcast (map-side) join: ship the small table to every mapper ----
broadcast = dict(users)                           # the broadcast collective, Ch 4
map_side = sorted((key, broadcast[key], payload)
                  for key, payload in events if key in broadcast)

print("reduce-side join rows :", len(reduce_side))
for row in reduce_side:
    print("   ", row)
print("\nbroadcast join rows   :", len(map_side))
print("matches reduce-side   :", map_side == reduce_side)

reduce-side join rows : 6
    ('u1', 'Ada', 'click')
    ('u1', 'Ada', 'purchase')
    ('u1', 'Ada', 'view')
    ('u2', 'Lin', 'click')
    ('u2', 'Lin', 'view')
    ('u3', 'Mat', 'click')

broadcast join rows   : 6
matches reduce-side   : True

4. Data Skew: the Dominant Failure Mode Advanced

Sort and join both rest on the same assumption: that the partitioner spreads keys evenly across reducers. Real data violates this assumption constantly. A handful of keys (a viral product, a bot account, a null placeholder, a default user id) appear orders of magnitude more often than the rest. When such a hot key hashes to one reducer, that reducer receives a disproportionate share of all rows while its peers sit nearly empty. Because the job finishes only when its slowest reducer finishes, one hot key turns a balanced plan into a single-machine bottleneck. This is precisely the straggler phenomenon analyzed in Chapter 2, here induced not by a slow machine but by an uneven data distribution: the machine is fine, its workload is not.

The standard remedy is salting. Append a small random suffix to the hot key so that what was one key becomes $S$ synthetic sub-keys, $\texttt{hot\#0}$ through $\texttt{hot\#}(S-1)$, which the partitioner scatters across up to $S$ reducers. The hot key's rows are then split into $S$ roughly equal streams instead of piling onto one machine. For an aggregation you finish with a second pass that re-combines the salted sub-keys; for a join you replicate the matching side across all $S$ salt values so every sub-stream can still find its partners. The cost is modest extra bookkeeping; the benefit is converting a single overloaded reducer into $S$ balanced ones. The code below makes a key 80 percent of the rows and measures the imbalance before and after salting.

random.seed(0)
N = 100_000
keys = (["hot"] * (N * 80 // 100)                                   # 80% one hot key
        + [f"k{random.randint(1, 5000)}" for _ in range(N * 20 // 100)])

R = 3
load = collections.Counter(hash(k) % R for k in keys)              # rows per reducer
print("rows per reducer (hash):", dict(sorted(load.items())))
print("busiest reducer share :", f"{max(load.values()) / N:.0%}")

S = 8                                                              # salt buckets
def salt(k):
    return f"{k}#{random.randint(0, S - 1)}" if k == "hot" else k  # spread the hot key

load_salted = collections.Counter(hash(salt(k)) % R for k in keys)
print("after salting hot key :", dict(sorted(load_salted.items())))
print("busiest reducer share :", f"{max(load_salted.values()) / N:.0%}")

rows per reducer (hash): {0: 86560, 1: 6878, 2: 6562}
busiest reducer share : 87%
after salting hot key : {0: 26680, 1: 26905, 2: 46415}
busiest reducer share : 46%

from pyspark.sql import functions as F

# Spark reads the small table's size and auto-selects a broadcast (map-side) join;
# the hint just makes the choice explicit. The large table is never shuffled.
joined = events.join(F.broadcast(users), on="user_id", how="inner")

# For a large-large join, Adaptive Query Execution detects and splits skewed
# partitions at runtime, no manual salting needed:
#   spark.sql.adaptive.enabled = true
#   spark.sql.adaptive.skewJoin.enabled = true

Sorting and joining complete the core distributed-algorithm toolkit of this chapter: you can now group, aggregate, order, and combine datasets too large for one machine, and you know the one failure mode (skew) that threatens all of them and the one fix (salting, automated or manual) that tames it. The next section turns to a different family of distributed computations, where the answer is not a reorganized dataset but a small extract or a numerical product: top-K selection, distributed matrix multiplication, and the PageRank iteration that ties them together. That tour begins in Section 6.6.