Distributed Sort & Hash

Concept. Distributed sort and hash are the two foundational data-movement primitives in a cluster: hash partitioning routes every row with the same key to the same machine (so joins and group-bys can run locally), and distributed sort orders the rows on each machine so per-key reductions can stream.

Intuition. When Spotify wants to count "lifetime listens per user" across a year of play logs, hash-partitioning by user_id sends every event for Mickey to the same node, every event for Minnie to another, so each node can sort its own subset and emit (user_id, count) pairs without ever talking to peers about other users.

Distributed Sort: One Total Order Across Machines

The first primitive puts the entire dataset in order across the cluster, in three passes: each machine sorts locally, the cluster samples split points, and rows redistribute so the machines line up end to end.

Three machines sort a dataset in three phases: each node sorts its own partition, the cluster samples two split points that cut the value range into thirds, then rows redistribute so node one holds the lowest third, node two the middle, node three the highest, a total order across the cluster.

Figure 1. Distributed sort in three phases. Each node first sorts its own partition locally. The cluster then samples the data to pick split points that divide the value range into equal pieces (here ≤3, 4 to 6, ≥7). Finally rows redistribute so each node owns one contiguous range, and reading the nodes in order gives a single total sort. Sampling is what keeps each node's share balanced even when the data is skewed.

Hash Partitioning: Foundation of Distribution

The Core Problem

When your dataset outgrows RAM, you split it across machines. Hash partitioning keeps related data together: it sends every row for one key to one machine, so a user's events never scatter between Ohio and Tokyo.

The Payoff: a Distributed Join

The point of keeping a key's rows together is that a join can then run locally. Hash both tables on the join key (here user_id), and every row for a given user_id, from either table, lands on the same machine. Each machine joins only its own slice, and the parts union into the full result.

Two tables, Listens and Users, are each hashed on the join key user_id, so all rows for user_id 7 land on machine one, 42 on machine two, 88 on machine three; each machine joins its own slice on user_id and the parts union into the full join, with no coordinator.

Figure 2. A distributed hash join. Both tables are hashed on the join key user_id, so every row for one user_id, from either table, is routed to the same machine (the colour is the key value: 7 blue, 42 orange, 88 purple). Each machine joins only the rows it holds, then the partial results union into the full join. No machine compares its rows against another machine's, so throughput scales with the cluster.

Consistent Hashing

The Problem: Adding/Removing Machines

Standard hashing (machine = hash(key) % N) reshuffles almost everything when you change the machine count:

  • From 3 to 4 machines: about 75% of data moves.

  • From 100 to 101 machines: about 99% of data moves.

The Hash Ring

Replace modulo with a ring:

  1. Create a ring: Numbers 0 to 2³²-1 form a circle.

  2. Place machines on the ring: hash(machine_id) determines their spot.

  3. Place keys on the ring: hash(key) finds their position.

  4. Key ownership rule: Move clockwise from a key until you hit a machine, that's its home.

A ring over 0 to 2^32 minus 1 with three grey nodes A, B, C and three grey key dots at hashed positions; from each key a pink arrow walks clockwise to the first node it meets, the node that owns it

Figure 3. Consistent hashing places both keys and machines on a ring over 0 to 2³²−1; each key belongs to the first node clockwise from it. A new node steals only the keys between it and its predecessor, about 1/N of them, so the cluster grows without the near-total reshuffle hash % N forces. DynamoDB, Cassandra, memcached pools, and CDN load balancers all run on this.

Key Takeaways

1. Hash Partitioning

The backbone of distributed processing

  • Keeps all rows with the same key on one machine.

  • Runs joins and group-bys in parallel without cross-talk.

  • Cost: 2N IOs (one read, one write).

  • Scales to hundreds of machines.

2. Distributed Hash Joins

Scaling joins

  • Route both tables by the join key.

  • Each machine joins its local partition.

  • Union the local results for the full join.

  • More machines give linear speedup.

3. Consistent Hashing

Scaling without a full reshuffle

  • Maps keys and nodes to a ring instead of modulo.

  • Adding a node only touches its immediate neighbors.

  • On average only K/N keys move, versus K keys with standard hashing.

  • Used by Cassandra, DynamoDB, Memcached.

4. Distributed Sort (BigSort)

Total order across the cluster

  • Phase 1: each machine sorts its own partition.

  • Phase 2: sample the data to pick global split points.

  • Phase 3: redistribute by split point and finalize.

  • The result is a total order across the cluster.