Why does cutting a 2 GB input into 50,000 pieces usually run slower than cutting it into 200?

## What a piece costs before it reads anything A **piece of the input** is one slice of the stored input that a single **worker thread** - one lane inside a worker process - reads and processes from start to finish. Cutting the input finer does not make the bytes cheaper to read; it multiplies everything that is charged *per piece* rather than per byte. In a staged runtime the per-piece charges are roughly: - **Creation and bookkeeping.** The **coordinating process** - the single process that turns the program into a graph, decides the pieces and hands them out - holds a record for every piece: its inputs, its assignment, its attempt number, its status. - **Dispatch and start-up.** Each piece is sent to a lane, set up there, and the lane reports back when it is done. - **Opening the input.** Each piece opens its source, seeks to its offset, reads headers or footers it needs, and closes again. On remote storage that is one or more network round trips whose cost is independent of how many bytes follow. - **Result collection.** Each completion returns a status and some metadata to the coordinating process, which must merge it into the run's state. - **Feeding a wide step.** A **wide step** is one a worker cannot finish from the records it already holds, because it needs records currently sitting on every other worker. Each upstream piece has to organise its records into one group per downstream piece, so the bookkeeping there grows as the product of the two counts. The byte-level cost of the exchange itself belongs to Moving Data Between Workers; only the multiplication is this leaf's concern. ## When the overhead outruns the work With 2 GB cut into 50,000 pieces, each piece holds about 40 KB. The fixed cost per piece is small but not zero - on the order of milliseconds in a staged runtime, and engines differ by an order of magnitude here - while reading and decoding 40 KB is far quicker than that. The run becomes mostly bookkeeping. | piece count | bytes per piece | what dominates | |---|---|---| | 20 | 100 MB | reading, but only a few lanes are busy | | 200 | 10 MB | reading; the fixed cost is noise | | 50,000 | 40 KB | the fixed cost, paid 50,000 times | ## The coordinating process is a serial section The second effect is worse than the arithmetic suggests. One process tracks every piece, so its bookkeeping, its heartbeats and its result handling are a serial section in an otherwise parallel run. Past some count - which depends entirely on the engine, and is the kind of number worth measuring rather than quoting - it cannot hand out work as fast as the lanes finish it, and the cluster idles while the coordinator catches up. A run in that state looks strange on a dashboard: enormous unit churn, low per-lane utilisation, and a wall clock that gets *worse* when you add machines. ## Long-lived lanes change the shape, not the moral All of the above assumes pieces are launched and torn down, which is the finite-job picture. In a **continuous job** - a run over an input with no end - the lanes are long-lived and there is no per-piece launch at all; the author sets a **declared operator width** instead, and it usually stands until the job is restarted. The analogous cost of setting that width too high is still real, but it is a different bill: - per-lane buffers and per-lane state overhead, paid for the life of the job; - network connections between producing lanes and consuming lanes, which grow as the product of the two widths; - fixed per-lane work at every recovery point, so a wider job takes longer to record one. So *do not over-divide* survives the move to a continuous runtime, but the reason changes from launch overhead to standing per-lane cost. ## What this is not - It is **not** the small-files problem. The file count this write leaves behind, and what those files cost the next job and the storage bill, belong to The Output Shape, and repairing them to lakehouse table compaction. - It is **not** skew - records concentrating on a few pieces so that they take far longer than the rest. Here every piece is equally tiny. - It is **not** an argument for the opposite extreme. Very few, very large pieces leave lanes idle, make each retry expensive and push a lane into **spilling**, which is writing part of its working set to local disk because it no longer fits in memory. ## The cheap correction When the count is already too high and the data is already divided, **merging pieces without moving records** - gluing neighbouring pieces together in place - cuts the count without a wide step. It is much cheaper than a full redistribution, and the catch is that it only merges neighbours, so each survivor is correspondingly larger and the merge cannot even out sizes.