Moving Data Between Workers

A redistribution costs bytes: what crosses the network, what the producing side writes to local disk where output is materialised, and processor time spent encoding it. Record counts predict none of those unless every record is the same width.

Sorting inside a piece orders only that piece. Unless the routing that filled the pieces gave each one a contiguous span of the key range, piece 3 can hold keys that belong before piece 2's, so reading the pieces in order interleaves ranges.

Records sharing a grouping key are combined where they were read, so each producing machine sends one partial result per key instead of every record. The collecting side combines partials into the same final answer, over far less traffic.

Matching records on a key normally forces every worker to send each record to whichever worker owns that key. Copying the whole small input to every machine instead lets the large input be matched where it already sits.

A step that needs records held by other workers cannot know it has them all until every piece of the producing step is done; computing sooner would publish a partial answer. That waiting line is a stage barrier.

Either load one landed piece into a keyed in-memory lookup and stream the other past it, or have both landed pieces arrive ordered by the key and advance through the two ordered runs together.

Draw a sample of the ordering key, pick boundary values from it so each piece receives one contiguous span of the range, redistribute records by comparing the key against those boundaries, sort each piece in parallel, and read the pieces in index order.

One states the per-key combining operation as part of the grouping step, so partial results can be formed on each producing machine. The other asks for the whole group first and computes afterwards, which obliges the job to deliver every record before anything can be combined.

The copy is paid once per worker, not once per job: two hundred workers means 2 GB sent two hundred times and 2 GB resident on two hundred machines at once. Cost tracks cluster width.

Each producing unit routes every record into one bucket per downstream destination; each collecting unit gathers its own bucket number from every producer that ran. Where buckets are written down first, collection can outlive the producer.

Up to eighty thousand — the product of the two widths. Each of the 400 producers holds one bucket for each of the 200 destinations, and each of the 200 collectors gathers one share from each of the 400 producers.

Compressing trades processor time for bytes on the wire, so it loses whenever the job is short of cores rather than bandwidth, or when the data is already compact, so the ratio approaches one and the encoding work buys almost nothing.

Neither does. The top few are found by keeping the best 100 inside each piece and merging those candidates, and ordering inside a key only needs equal keys to land together, which any routing on that key already gives. Range boundaries serve a complete result read in order.

Folding replaces the records sharing a key with one partial result, so its saving is the records-per-key ratio. When keys are near-unique that ratio is about one: every record becomes its own partial, and the traffic is unchanged while the bookkeeping is added.

Collecting bytes from producers that have already finished may overlap producers that are still running; computing on them may not, because the consumer's input is not complete until the last producing piece is done. Transfer overlaps the wait, compute does not.

Because its producers never finish. A line across a job says compute waits until every upstream piece is done, and over an endless input that moment never arrives, so operators must emit from partial input instead of waiting for completeness.

The keyed lookup's resident set is the whole loaded side, whatever the distribution of keys inside it, so it is unchanged. The ordered walk's is the run sharing the key being matched, so one enormous key is exactly what inflates it.

The keyed in-memory lookup cannot hold it: runtimes respond by cutting both landed pieces further on the key and matching sub-pair by sub-pair, by switching to the ordered walk, or by failing. The walk instead pays extra ordered passes over local disk.

Width multiplies every byte of the movement after it: 200 million records at 40 KB is about 8 TB on the wire, against 12.8 GB for the grouped fields alone. The repair is to widen after the movement, not before.

The sample misrepresented the key distribution, so the cut points do not split the data evenly. The output is still correctly ordered; what suffers is the run, which is now bounded by the one oversized unit of work and its memory pressure.

Add an explicit first step that walks each piece of the input, accumulates per key in a bounded table, and emits one partial per key; then redistribute and combine the partials. It costs producer memory, a lookup per record, and a two-step shape a later reader must understand.

What it becomes in memory, not what it occupies on disk. Decoded into records and indexed for lookup, a compressed file commonly expands several-fold, and that expanded figure is what every worker must hold at once.

Wherever the copy is assembled or held: on engines that gather it centrally, the gathering process can exhaust its memory first, and otherwise every worker fails at nearly the same point, because all of them are doing the identical thing.

Each line lifts when its slowest producing piece finishes, not when the average one does. If every piece was already running at once, extra workers add capacity nobody was waiting for, so the job's wall clock stays roughly the sum of three maxima.

The producing work again, not just a transfer. Those buckets are intermediate bytes on one machine's local disks, usually in a single copy, so nothing can be re-read: engines that kept a description of how the piece was made re-run it.

Nothing durable. The routing rule still picks one destination per record, but the bucket is only a send buffer flushed over a channel to a consumer that is already running. There is no stored share to request, and none to request twice.