Processing Engine Concepts

A bounded input has a last record the reader can reach; an unbounded one does not. Reaching that last record is what lets a job produce one total, one global sort or one rank, call it final, and stop.

Before the first read, a cluster must be granted capacity, launch a worker process on each machine, copy the program and its libraries to each, build a plan and list the input. None of that touches data.

In a workflow scheduler's graph an edge means only 'start after the previous step reported success'; nothing travels along it. Inside one program handed to a cluster execution engine, an edge carries the records themselves from one step to the next.

A cluster execution engine runs your program on machines that are part of your deployment, over storage you point it at, along a plan your code shapes. A managed query service owns all three and shows you none of them.

A fixed reporting cadence usually means the runtime executes repeated small finite runs: it collects whatever arrived in an interval, runs an ordinary finite job over that slice, and repeats. The floor is the interval plus that run.

A named operator carries its meaning - the engine knows it keeps rows, or groups by a key - so it may reorder, narrow, fuse or skip it. A handed-over body means only 'call this per record', so it runs exactly where it stands.

Nothing in the job asked for an order, so none was kept. The work is cut into pieces processed by many workers at once, and rows appear in whatever sequence those pieces finished and were collected. Only a declared ordering step makes row order stable.

A describing call only adds a step to the graph the engine will run and computes nothing. A demanding call asks for an answer or for output to be written, and that demand is what makes the recorded steps execute.

Two: read-time column narrowing, which tells the read to produce only the two columns later steps use, and filter at the read, which moves the date condition down so rejected rows are never produced. Both shrink what enters the job.

Four: the points where records must be sent between workers, where the listing breaks into runs of steps between those points, how many pieces each step processes at once, and what the read was told to skip.

The program is kilobytes and the input is gigabytes, so shipping the work to the bytes is far cheaper than pulling the bytes across a shared network to wherever a thread happens to be free.

One. A stored form that cannot be decoded from an arbitrary offset yields exactly one piece of the input — one slice a single worker thread reads end to end — so every other thread in the cluster stays idle during that read.

Dependency shape decides: if each output depends only on records already in the thread's piece, it finishes alone — a narrow step. If one output needs records spread across every piece, the step is wide.

Eight. The piece count is the ceiling on how many worker threads can be busy; the machine count only decides how many threads exist, so the other 192 lanes stay idle.

Roughly one file per piece, so about 2,000. Each worker thread writes the records it already holds, where it is; nothing gathers them first. The piece count at the final step, not the machine count, set the file count.

Both inputs are cut on the matching key by the same routing rule into the same number of destinations and sent across the cluster, so records sharing a key value land on one worker, which then matches them locally.

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.

The median shows the typical piece finished in about a second while the maximum shows one ran nearly the full 52 minutes: a long tail of one or two outsized pieces, not a step that is uniformly slow.

A key-to-destination rule sends every record carrying the same key to one destination, and that mapping is fixed. Extra machines add destinations, not a way to split one key, so the commonest key still runs on a single worker.

Appending a small integer to the grouping key turns one key value into several, so its records reach several destinations instead of one. A second pass then re-groups those partial results under the original key to produce the real answer.

That is a straggler: a unit whose slowness comes from the machine under it - a failing disk, a busy neighbour process, an empty local cache - rather than from holding more records than its peers.

Runtime replanning - re-deciding unrun steps from the sizes a finished step actually produced - can change only what has not started: the shape of the next step's pieces, or a join switched to copying the small side. Finished work stands.

Pinning tells the engine to keep the computed intermediate after the first reader finishes, so the second reader reads the kept copy instead of re-running every step that produced it. Without a pin, that branch is computed twice.

Memory is owned per worker process, so a failure inside one unit of work says nothing about total cluster capacity. Idle machines cannot lend memory to the process that is short; what reached that one unit has to change instead.

Each field can become a separate object carrying the runtime's own per-object bookkeeping - a header, alignment padding, and a pointer to it from the record. A record of many small fields therefore costs a multiple of its raw bytes, not a small addition.

Spilling is planned degradation: an operator that cannot hold its working set writes part of it to disk attached to the worker and reads it back to finish. The answer is identical; only the runtime grows.

Operators borrowing room while they run, any results the job asked the engine to keep, and runtime overhead the engine never counts - all shared by every unit of work inside that one process, with no lending from idle workers.

A completeness claim - a watermark - is a timestamp the job carries with the records, asserting no record older than it is still expected. It makes any group ending at or before it eligible to close.

Nowhere visible: a record whose group has already closed is usually discarded without an error, so the totals are simply short. An explicit count of records dropped for lateness is what makes the loss visible at all.

Out-of-order arrival is a property of the transport, not a fault: buffered devices, retries and parallel inputs merged together all deliver an earlier-stamped record after a later one. It costs nothing by itself until a time-based grouping has to be treated as finished.

Three moments compete: when the order was placed, stamped in the payload; when the receiving system wrote the record down; and the wall clock of the worker that handled it. A daily count belongs to the payload moment.

Nothing in an endless input reports what has not yet been sent. The job infers the claim from the moments it has seen plus a standing assumption about how far behind a record may run, so it can be too early or too late.

A running total needs one number per key while the group is open. A median needs every value in the group, because no single carried value can be merged into the answer. Same interval, very different memory.

The record's own moment supplies the bound: it selects one version of the reference row, the one whose span of validity contains that moment. Nothing accumulates and no group closes, so each record yields its output once its match can be resolved.

Neither input ends, so a record on either side may meet its partner later. Both sides are therefore held while the other is awaited, and a bound on how far apart their two moments may be is what lets a held record ever be dropped.

Five. A span of fixed length restarted every step shorter than that length overlaps its neighbours, so a record's moment falls inside length-divided-by-step spans at once and is counted in every one of them.

Three shapes: back-to-back spans of equal length, where a record lands in exactly one group; fixed-length spans restarted every shorter step, where a record lands in several at once; and groups the data itself closes after a stated quiet period.

New code has to read back the retained set the previous build wrote — everything the job holds between records. A stateless redeploy discards nothing; here, entries the new build cannot match or decode mean a refused start, or history silently lost.

Key-bound state is addressable only under the grouping key of the record in hand. Each key's entry lives with the worker that owns that key, and the step is handed no way to reach another key's entry.

A deduplication check only ever adds: each new id creates an entry and nothing deletes one, so the retained set grows with distinct ids for the life of the job. Only a removal rule bounds it.

The running total and the repeat-identifier dropper hold entries between records: one accumulator per user, one entry per identifier already seen. The filter holds nothing, because its verdict depends only on the record in hand.

The cap is the memory of that one worker process, shared with everything else the process is doing. There is normally no overflow path, so passing it degrades and then kills the worker rather than quietly moving entries to disk.

Resuming restores each worker's accumulated values and the recorded read position exactly as they stood at one coherent moment. Everything the job did in the following 40 seconds is read again and recomputed.

Recovery is reprocessing: a restart re-reads input the job had already handled and writes it a second time. The destination remembers nothing of the first attempt, so a plain append adds those rows again.

The engine re-runs only the failed unit of work — the smallest piece it hands one worker — on whichever machine has capacity. That is correct only when the piece can be recomputed from re-readable input and its partial output is discarded.

Recovery is reprocessing: a restart re-reads input it already consumed. So the source must retain those records and let a reader resume from a durably recorded position — a source that forgets each record after handing it over makes the lost work unrecoverable.

A marker injected at the sources travels with the records; each worker saves its own part as the marker reaches it and then passes it on, so every part covers the same prefix of the input.

A standing pool that is already up and shared by many jobs; a cluster raised for one job and torn down with it; or a managed compute service that shows no machine at all. Sizing moves from submitter to provider across the three.

On pieces of work that have not started yet, normally from the next round of work onward. A piece already running on another worker process is not cut in half and shared, and finished work is not redone to spread it more evenly.

Everything that machine held goes at once: every worker process on it, the pieces they were running, anything cached there, and intermediate output produced there that other workers had not yet fetched. Finished work can therefore have to be produced again.

Work slots in one worker process share that process's single memory pool, its local scratch space, its fixed start-up overhead and its fate: they compete for the same memory, and all of them die together when the process does.

Nothing starts until machines free up. Submitted jobs wait in an admission queue, and a share policy — strict order of arrival, weighted shares, or a guaranteed minimum per group — decides which waiting job the freed capacity goes to.

Backpressure: a step further along the job graph cannot accept work as fast as this one produces it, so this one blocks. The figure locates a bottleneck ahead of the resisting step, not a fault inside it.

A worker's diagnostic lines are written on the machine that ran it, and the cluster releases that machine when the work ends. Only what the run carried off its machines stays readable.

Four million records behind is a level, and recovery depends on the trend. Compare two readings taken minutes apart: a distance that is growing means the job is losing ground, a shrinking one means it is already draining.

Five numbers cover almost every 'slow' report: processed throughput in records and in bytes per second, duration per step of the job graph, distance behind the input, and failed or retried units of work. Read them before changing anything.

Anything the code reads from the present: the current time written into output, timeouts and inactivity gaps measured in real seconds, rate thresholds per minute, expiry of held state, and relative date filters. Compressed into one hour, none behaves as it did live.

The first rows of a file are whatever was written first - one day, one source, one writer's share - so they carry neither the whole input's key distribution nor the rare record shapes that actually break the job.

The per-record rule can: a plain function taking ordinary values and returning ordinary values. It stops being callable once its signature mentions a runtime type, or it reads configuration, storage or the clock itself instead of taking them as arguments.

Sleeping proves only that the machine's wall clock advanced. Instead hand the job records stamped with chosen moments and advance from the test the job's own claim that nothing older will arrive, so groups close on command.

A distributed run promises the right records, not an order, a file layout or bit-identical arithmetic. A line-by-line comparison asserts all three at once, so it fails on arrangement while every value is correct.

Running both versions over the same production input, with only one publishing, measures the change against real volume, real key distribution and real dirty records — evidence a test cannot give, because a test holds only the cases its author imagined.