Scenarios

n × T × λ = 20,000 × 0.05 h × 0.002 = 2. An attempt then succeeds with probability e⁻² ≈ 14%, so the expected number of full attempts is e² ≈ 7.4. If one attempt takes an hour, that is over 7 hours of cluster time, and the delivery time is unpredictable.

Materialize the edges between stages, so a task is the unit of failure: each failure then costs one 3-minute task.

Writing and reading the intermediate data, and a barrier between stages that stops computation overlapping communication. Keep streamed edges for short dashboard queries, where n × T × λ is far below 1.

Tasks must be deterministic and idempotent for retries to be safe.

For a billion rows, the cost of crossing between operators for every row (tens of cycles each) dominates, so it must be amortized or eliminated. A compiler adds a fixed cost per query: with 50 ms to compile and 20 ns saved per row, it breaks even at 2.5 million rows. Small dashboard queries fall below that, so a compiler-only engine would be slower on them, and you would need tiering. A vectorized engine with batches of about 1,024 values has no such threshold, and keeps many cache misses in flight on memory-bound scans.

Whether the workload is compute-bound (many expressions) or memory-bound (large hash tables).

Push-based and vectorized, with hand-written primitives: easier to profile, debug and port. Add compilation later only for the few computation-heavy pipelines, compiling in the background and switching when ready.

The performance gap between the two designs is smaller than marketing suggests; engineering cost decides.

Each iteration needs a consistent snapshot of the previous one, so bulk synchronous processing (BSP) fits, and it is deterministic.

Each superstep costs the slowest computation plus the slowest communication plus barrier latency. With 200 workers and a coefficient of variation of 0.2, the straggler penalty is about 1 + 0.2 × √(2 ln 200) ≈ 1.65 per superstep, paid 50 times with no averaging. And computation cannot overlap communication across a barrier.

Balance partitions, and send messages in a pipeline within each superstep. If the algorithm tolerates stale values, asynchronous or bounded-staleness updates help, at the cost of determinism and reproducibility.

BSP first; measure the penalty per round; go asynchronous only if the penalty, multiplied fifty times, is unacceptable and convergence is robust to staleness.

1,000 producers × 1,000 consumers = 10⁶ shuffle objects of 100 KB. At $5 per million writes, that is $5 per query in writes ($0.05 per GB), about the same again in reads, plus a latency floor of 10–30 ms per object.

Functions cost about 3–8× as much as virtual machines per unit of work. But with a few queries a day, a cluster would sit idle most of the time, so paying 5× for the seconds you use beats paying 1× for the hours you do not.

Requests grow as M × R, so doubling the width quadruples request cost; cold starts; the maximum execution time.

A tree-structured exchange, fewer and larger partitions (tasks of seconds, not milliseconds), or a fast temporary store for intermediates.

Yes for bursty work with narrow shuffles; no for steady heavy load. Re-check the price ratio regularly: it is a pricing decision and has changed before.

A push boundary with no way to refuse: the producer decides when data moves, so the consumer cannot say "not yet".

Bound the queue and make the producer block, or use credits as Flink does. Size the buffer by Little's Law: depth ≈ arrival rate × the consumer's latency, enough to keep the consumer busy through the producer's pauses and no more. Too small and the pipeline serializes; too large and you spend memory hiding variance.

The producer now stalls, so throughput is limited by the slow stage, as it always really was, and LIMIT or cancellation needs a stop signal.

Make the boundary pull-based: refusal comes free, at the cost of a call per unit, which batching can pay back.

From per-task counters, find f, the largest partition's share of the work. If it is 5%, the speedup is capped at 1 / f = 20, whatever the cluster size.

Plot task duration against input bytes to tell skew from stragglers and imbalance.

Split the unit: salt known hot keys; split partitions adaptively if shuffle files can be read by range; or separate heavy hitters, if a sketch can find them.

Speculative execution: a copy of a big task is just as big.

The number of stages: pipeline breakers fix how much can overlap.

Salting multiplies the other side of a join by the number of salts; adaptive splitting needs sorted, indexed shuffle files.

Each block would be 10 TB ÷ 10⁸ = 100 KB, far below object storage's efficient floor of 2–8 MB. Requests at 100 KB per object cost about $0.05 per GB, about $500 for 10 TB. The width ceiling is √(10 TB / 2 MB) ≈ 2,236, so 10,000 is well above it.

Reduce the bytes first (projection, pre-combining, broadcasting any small side), reduce the width towards 2,000, use sort-based shuffle, and put a push-based shuffle service with local disks in front of object storage, so consumers make large reads.

An extra hop and a shared stateful service; keep the unmerged blocks as a fallback.

Scaling out mid-query pays only if remaining work / p − remaining work / p′ exceeds the startup latency. With 60 core-seconds of work left, going from 10 workers to 20 saves 3 s and costs 30 s of startup.

The job may already be at its task-count ceiling (total work ÷ the smallest sensible task).

Elasticity belongs between queries, or early in long ones. For elasticity inside a query, over-partition and detect the need early.

Shuffle compression was turned on when it paid, when codec throughput exceeded BW × r / (r − 1). A 1 GB/s codec with ratio 2 would now need 25 GB/s to win on a 12.5 GB/s link. Turn it off, keeping it for slow cross-zone links.

A non-deterministic partition function or UDF, for example a random salt, sends rows to different partitions on a retry. Make it deterministic (a hash of the task ID and row index) before enabling retries or speculation.

That output from failed attempts is never read: only one attempt's output should be committed for each task.