How a Partitioner Maps Keys to Reduce Tasks

Grouping by key across a cluster requires a rule that sends all records of one key to one place. That rule is the partitioner, and the number it maps into, the reduce task count, has to be chosen before the first record is read. Everything expensive about a shuffle follows from that ordering constraint.

The invariant that makes reduce-side work correct

Reduce-side operations — aggregation, duplicate removal, sorting, joining — all assume that every record with a given key is presented to the same task. The partitioner supplies that assumption: it is a function from a record to an integer in the range zero to $R-1$, and record $r$ is written to output partition $p(r)$ by the map task that produced it, then delivered to reduce task $p(r)$.

Nothing else is required of the function. It does not have to be balanced, ordered, or stable across reruns for correctness of a single attempt, but it does have to be deterministic: a retried map task must place the same key in the same partition, or the reducer will aggregate half a key’s records and report a wrong answer. Details of that failure belong with shuffle fault tolerance.

$R$ is a job-level setting, not a derived value. The reference MapReduce implementation historically defaulted to a single reducer, which serializes the entire reduce stage; parallel frameworks default to a fixed number unrelated to input size. In Spark SQL the long-standing default is 200 partitions whether the input is one gigabyte or one hundred terabytes.

Hash modulo and its consequences

The usual partitioner computes $p = \lvert h(k) \rvert \bmod R$, where $h$ is a hash of the key. Three properties fall out of that arithmetic.

The first is that the mapping is key-based, so equal keys always collide into the same partition and unequal keys may do so. That is the desired behaviour: a partition is not a bucket of one key but a group of keys processed together.

The second is that the hash must be well mixed and the sign convention handled. Java’s hashCode returns a signed 32-bit integer, so roughly half of all keys produce a negative value; an implementation that applies modulo to the raw value can produce a negative partition index and an out-of-range segment. Masking the sign bit before the modulo is the standard fix. Poor hashing is a separate and subtler failure: keys that are sequential integers hash to sequential integers, and sequential integers modulo a small $R$ distribute evenly only if the sequence is dense and longer than $R$, which makes sparse or clustered key ranges unpredictable.

The third property is statistical, and it is the one people misjudge. For $n$ records spread over $R$ partitions by a well-mixed hash, the expected count per partition is $n/R$ and the relative deviation falls as $1/\sqrt{n/R}$. A partition holding a million records deviates from the mean by about a tenth of a percent; a partition holding a hundred records deviates by tens of percent. Uniform distribution therefore becomes a good approximation only once partitions hold thousands of records, which is an argument for choosing $R$ from the data volume rather than from habit. When keys themselves are not uniform — a head-heavy distribution where a few keys dominate — the balance argument collapses entirely, and the remedies are the subject of hot-key skew mitigation.

Because $R$ appears inside the modulo, it is part of the partition function rather than a deployment detail. Moving from 200 partitions to 201 sends a key with hash $h$ to $h \bmod 200$ in one plan and $h \bmod 201$ in the other; for most keys those values differ, so nearly every record moves. Repartitioning is a full shuffle of the data, never an incremental one, which is why the count cannot be tuned after the map stage has run and why frameworks that want to resize partitions later must do it at the stage boundary, as adaptive partitioning at runtime does.

Range partitioners and total order

Hash partitioning discards order. Each partition is internally ordered if the shuffle implementation sorts, but nothing orders partition 3 relative to partition 4, so concatenating the outputs gives arbitrary key order. A range partitioner assigns by comparison against boundaries instead of by hash: it samples the input, chooses $R-1$ split points, and sends a record to the partition whose interval brackets its key. Partition zero then holds the smallest keys and partition $R-1$ the largest, so reading partitions in order yields a fully sorted stream, and a downstream merge join or range scan can walk it sequentially.

The costs are real. Range partitioning needs a sample before it can place anything, so it either adds a preparatory pass or relies on reservoir sampling inside the first stage. Bad boundaries concentrate load: if the sample misses a heavy region of the key distribution, one partition absorbs it, and unlike hash partitioning there is no statistical averaging to fall back on. It also requires an ordering that all participants agree on; collation and numeric type coercion must be identical on both sides or records land in intervals that a reader would not predict.

Pig Latin’s designers made range partitioning the default for its sort and join operators precisely because those operators need order, while group and cogroup accept either. The choice of partitioner is therefore a statement about which operations the stage must support, not a performance knob.

Choosing the count

Too few partitions produce slices larger than a task can hold in memory. At 10 TiB with 200 partitions, each task faces roughly 50 GiB of input, far beyond any in-memory merge, so the reduce side degenerates into repeated disk passes. Too few partitions also concentrate the shuffle: the stage finishes when its largest partition finishes, so a count far below the slot count wastes the cluster.

Too many partitions multiply metadata and request overhead. Each map task writes one segment per partition, so a job with 1,000 map tasks and 200 reduce tasks already creates 200,000 intermediate segments for the coordinator to track and for reducers to fetch individually. Raising the reduce count to 20,000 turns that into 20 million segments, each of them a small read with its own per-request cost. The fetch fan-in per reduce task is exactly the number of map tasks, so a very large partition count and a very large task count multiply rather than trade off.

Empty partitions are a related waste. If the data has fewer distinct keys than $R$, some partitions receive nothing, yet their tasks are still scheduled, started, and completed. A job with 1,000 reduce tasks over data with 50 distinct keys spends 950 task slots producing empty output.

The count also decides file sizes downstream. Each partition becomes a file in the stage output, so $R$ sets the output file count, which feeds the small-file problem back into input splits for whatever job consumes that output next. A schedule of daily jobs each emitting 50,000 tiny files compounds the metadata pressure that split sizing was meant to control.

The per-record cost of the function itself

A partitioner runs once per record emitted by the map function, which puts its cost on the critical path of the whole job. Hashing a string requires scanning its bytes; hashing an already-hashed key costs another function call per record; hashing a composite key requires serializing its components in a fixed order, and if the serialization used by the partitioner differs from the one used elsewhere, the same logical key can land in two different partitions. Cryptographic hashes mix better than the identity hash most languages provide, but they cost an order of magnitude more per byte, and the modulo that follows already discards most of the extra mixing.

The modulo needs care in two situations. With a partition count that is a power of two, the result depends only on the low bits of the hash, so a weak hash whose low bits are poorly distributed leaves some partitions empty and gives others twice their share — the classic failure of hashing integers that differ only in their high bits. And when the hash range is not an exact multiple of the count, the remainder is marginally biased; for a 32-bit hash and 200 partitions the imbalance is on the order of one part in ten million, which is irrelevant next to the imbalance introduced by a skewed key distribution.