Shuffle compression, serialization and determinism

When does compressing shuffle data pay, and what must be deterministic for retries to be safe?

Shuffle compression: a throughput comparison, not a ratio comparison

Uncompressed time to ship S bytes: S / BW, where BW is the link bandwidth.
Compressed time: S / C + (S / r) / BW, for a codec with throughput C and compression ratio r. (First compress at rate C, then send S/r bytes.)

Compression wins if and only if:

This is encoding's compression break-even (D/B > r/(r − 1)) with the network link in place of the disk, so C > BW · r/(r − 1).

For r = 2, the codec must be twice as fast as the link. For r = 4, only 1.33× as fast (4/3).
LZ4 and Snappy (fast codecs) run at several hundred MB/s to about 1 GB/s per core.
1 Gbps link (125 MB/s): any of them wins by a wide margin (at r = 2 the codec must beat 250 MB/s).
10 Gbps (1.25 GB/s): LZ4 at r = 2 needs 1.25 × 2 = 2.5 GB/s per core and loses.
100 Gbps (12.5 GB/s): nothing wins.

The compression decision inverted as networks got faster. This is the "fixed cost per operation" idea with its migration clause: when hardware gets faster, the break-even moves. Shuffle compression was clearly right in 2010 and is workload-dependent today. It is one of the few places where such a reversal can be checked against a date.

Why shuffle codecs are different from storage codecs. Stored data is written once and read many times, so a slow, strong codec spreads its cost over every later read. Shuffle data is written once and read once. There is nothing to spread the cost over, so the best choice moves to the fast and weak end of the codec range. The ratio matters only through the r/(r − 1) term, which approaches 1 as r grows (for r = 3 it is 1.5, for r = 10 it is 1.11). That is why shuffle codecs and storage codecs are different codecs. It is a derivation, not a convention.

Serialization

A generic, reflective serializer (one that discovers an object's fields at run time) costs hundreds of nanoseconds to microseconds per row. That is paid per row, at every shuffle. At 10⁹ rows and 500 ns per row: 10⁹ × 500 ns = 500 s, minutes of pure CPU.

The convergent fix is to make the wire format be the engine's own internal format, so serialization turns into a framed memory copy. Two forms: compact binary row layouts, or columnar batches shipped as they are.

generic:   for each row: for each field: reflect -> type switch -> encode   # per-row cost
internal:  copy batch buffer + small header to the socket                   # per-batch cost

This is the same insight as in Inside one worker, applied to another boundary: remove a per-tuple cost by choosing a representation that needs no per-tuple work. The same move at two different boundaries is evidence that the per-tuple-boundary problem is one problem, not several.

One subtlety that causes trouble. Columnar shuffle with dictionary encoding (replace each distinct value by a small integer, with a table that maps integers back to values) must keep encoder state per target, because each consumer receives a different subset of values. So encoder state is multiplied by R. That brings back exactly the R-fold multiplication that sort shuffle removed from the buffers. Dictionary-encoded shuffle is therefore practical only with a modest R, or with per-batch dictionaries that sacrifice compression ratio.

Fault tolerance and determinism

The shuffle is the plan's only re-readable point, so it is the recovery substrate whether or not it was designed as one. Two requirements follow. Both are correctness requirements, not performance requirements.

The partition function must be deterministic

A retried producer that assigns rows to different consumers than the first attempt did sends rows to different places. Some rows are then duplicated into one partition and lost from another, silently, with no error anywhere. That is why a non-deterministic key, or a non-deterministic function written by a user, upstream of a shuffle is a correctness hazard. It is the same hazard that speculative execution created in Where parallelism comes from, arriving from a second direction.

# DANGEROUS: different on every attempt
partition = hash(row.id) % R             # fine if hash is stable
partition = random.randrange(R)           # NOT safe to retry: attempt 2 differs from attempt 1
partition = hash(row.key + now()) % R     # NOT safe: depends on the clock
# SAFE alternative for "random" balancing: seed from stable data
partition = hash(task_id, row_index) % R  # same on every attempt (if row order is stable)

(Python's built-in hash for strings can differ between processes unless seeded; use a stable hash function in an engine.)

Row order within a partition must be deterministic

It must be deterministic if any consumer is order-sensitive, or if partial output from a failed attempt was already consumed. Making the output deterministic by construction is the only cheap answer. The alternative is to make consumption transactional.

The collector problem

Shuffle outputs are an immutable substrate, so they grow an indirection layer and a collector:

the map-output tracker: a registry that maps each partition to its locations and offsets;
the cleaner: shuffle cleanup at job end, or reclamation by time-out in an external service.

The failure mode is the usual one for such designs, and it is recognizable in practice: the cleaner falls behind under exactly the conditions that make the most garbage: many short jobs, long-lived executors and cancelled queries. The resulting disk exhaustion looks nothing like its cause.

Case, resolvedConstructed case · Hana at Gridlark

A faster network made compression slower

Mateo answers the three symptoms one by one.

Compression slower after the upgrade. At 100 Gbps (12.5 GB/s per link), a codec at about 1 GB/s per core with r = 2 needs more than 25 GB/s to win. It cannot. The configuration was written for 1 Gbps links. Compression is now off for shuffle on the new links (and kept on for the slow cross-zone links where C > BW·r/(r−1) still holds).
The total off by a few thousand rows. A user-defined function upstream of the shuffle used a random number to spread rows ("salt"), and the retried task drew different numbers than the first attempt. Rows moved to other partitions: some were counted twice, some were lost. He replaces the random draw by a hash of (task id, row index), which is the same on every attempt, and turns on a check that the partition function is deterministic before retries and speculation are allowed.
The disks full of dead shuffle files. Cancelled queries never reached the cleanup step, and the cleaner's time-out was one week. He ties the cleanup to query cancellation and shortens the time-out. The map-output registry now drops entries when the query ends.

She leaves the meeting with the sentence that Mateo wrote on the whiteboard at the beginning of the week: the plan says what work exists; the runtime decides how much of the machine is idle.