Compute Near the Data

## What "near the data" actually means A **piece of the input** is one slice of the stored input that a single **worker thread** — one lane inside a worker process, which is itself one operating-system process on one machine — reads and processes from start to finish. Before that thread can compute anything, the slice's bytes have to be in its memory, and there are only two routes: - **Local**: the bytes already sit on a disk attached to the machine the thread runs on, and the read never leaves the box. - **Remote**: the bytes sit on some other machine, and they cross a network link that every other machine in the cluster is also using. Placing compute near the data is the scheduling preference that tries to produce the first case. The **coordinating process** — the single process that turns the program into a graph, decides the pieces and hands them out — asks where each slice's bytes live and tries to hand that slice to a worker thread on one of those machines. ## The arrangement that made it pay The idea was invented for a **cluster file system**: storage running on the same machines as the compute, where each block of a file is written to the local disks of a small number of those machines. That arrangement is what gives a scheduler a choice worth making — there exists a machine that *has* the bytes, so preferring it means something. On storage that no compute machine holds, the preference has nothing to express at all. Against that background the arithmetic is one-sided: | what moves | typical size | how often | |---|---|---| | the program | kilobytes to a few megabytes | once, to each machine that runs part of it | | the input | gigabytes to terabytes | every byte, every run | Shipping a small program to where the large data already sits costs almost nothing. Shipping the data to wherever a lane is free costs the whole input, over a path shared with every other piece of work in the cluster. The old slogan — *move the computation, not the data* — is a claim about that ratio, not a law. ## What the saving is, precisely Interviewers push here, and the precise answer is worth having: 1. **It is the gap between a local read and a remote read, and nothing more.** If local disk and the network deliver comparable bandwidth for this input, a local placement saves nothing you can measure. 2. **It is per-byte and per-run.** A run reading ten terabytes benefits ten times as much as one reading one terabyte; the program shipped in the other direction is the same size either way. 3. **It is congestion relief as much as latency.** Hundreds of threads reading remotely converge on the same switches. Local reads keep that traffic off the shared path entirely, which helps every other job on the cluster too. 4. **It applies to reading the input, and only to that.** Once the program reaches a step a worker cannot finish from the records it already holds, records have to be redistributed between workers whatever the placement was. These are two different costs: reading remotely happens *before* the work, redistribution happens *because of* it, and placement only influences the first. ## What it is not - It is **not** a copy step. Nothing relocates bytes to a chosen machine before the run; the scheduler chooses among machines that already hold them, or gives up and reads across the network. - It is **not** a correctness mechanism. A remote read returns the same bytes; only the cost differs. - It is **not** the same word as the on-disk grouping of files into directories named for a column's value. That grouping decides which files are read at all; placement decides which machine reads them. ## Where engines genuinely differ Do not state one behaviour for the whole class: - Some runtimes model an explicit ordered preference and will briefly hold a piece hoping for a nearer machine; others express no preference and place purely by which lane is free. - In a **finite job** — a run over an input that ends, so the bytes can be measured before it starts — the pieces are derived from stored data and can be matched to the machines holding it. In a **continuous job** — a run over an input with no end — there is nothing stored to match against; operator placement is settled when the job starts rather than re-decided per record. - Where the input lives in an **object store** — remote blob storage reached over the network — no machine holds the bytes, every candidate is equidistant, and the preference is simply absent. That is the common modern deployment. ## The answer that lands Name the three conditions the optimisation needs: compute sharing machines with storage, an input far larger than the program, and a network materially slower than local disk. Someone who can name those has also explained, without being asked, why most deployments today get nothing from it.