Summary and key takeaways

The three idlenesses

Idleness	Waiting for	Typical cure	What the cure costs
Dependency	The producer	Buffers, pipelining, batching	Memory (Little's Law: items in flight = rate × time)
Imbalance	The slowest peer at a barrier	Smaller units, stealing, splitting, speculation	Dispatch overhead, duplicate work
Exclusion	A shared resource	Partitioning, shuffle	An O(M × R) edge count

The two numbers to carry. Bytes per shuffle connection b = S / (M × R), compared with the substrate's minimum efficient transfer. And the barrier penalty E[max]/E[mean] ≈ 1 + c·√(2 ln n), paid again in every round.

Mateo and Hana's lessons, in order:

The idle dashboard (Three idlenesses). An optimal plan can still leave 75% of the cores idle. Name which idleness dominates before buying machines.
The 40× slower loop (Inside one worker). Cost sits at boundaries (a call per tuple). Direction (pull or push), granularity (tuple, batch, stage) and form (interpreted or compiled) are independent choices. Batch about L1 ÷ row width, about 1,024. Compilation breaks even at 50 ms ÷ 20 ns = 2.5 million tuples.
The failing nightly job (Across machines). Across machines a task must last at least about 1 s (10 ms dispatch, 1% target). n × T × λ decides stream or materialize. BSP pays the straggler penalty every superstep; actors need bounded mailboxes; serverless is priced in requests.
Doubling the cluster (Where parallelism comes from). T_p ≥ max(T₁/p, T∞). The largest partition caps speedup at 1/f, and under Zipf-1 that is about ln(distinct keys). Pipeline breakers define stages; decomposability gates intra-operator parallelism; work stealing reaches the bound but cannot split a task; speculation helps stragglers only when the cluster has room.
The 7% key (Skew and elasticity). Classify the tail (duration against bytes) before choosing a fix. Skew needs the unit split: salting, adaptive splitting, heavy-hitter separation. Join skew multiplies (L × R). Elasticity is limited by work ÷ minimum task, and mid-query scale-out pays only when remaining work ≫ acquisition latency.
Too many open files (The shuffle). A shuffle is a distributed transpose. Width ceiling M = R ≤ √(S/b_min). Sort shuffle replaces R files with R offsets; consolidation collapses the M axis; push-based and remote services merge across producers (and win mainly through elasticity); trees need associativity; broadcast when R × S_small < S_large.
The faster network (The shuffle). Compress only if C > BW × r/(r−1); serialize as a copy of the engine's own format; partition functions must be deterministic; immutable shuffle output needs a registry and a cleaner that keeps up.

Decision procedures in one table:

Question	First check	Rule
Why is speedup flat?	Largest partition share f	Ceiling is 1/f; stop buying cores there
Which fix for the long tail?	Duration against bytes scatter	Spread: finer tasks or stealing. One slow task: speculate. Duration ∝ bytes: split the unit
Static or dynamic scheduling?	Are stage edges streamed?	Streamed means static; materialized allows per-stage adaptation
How small can tasks be?	Dispatch cost F	Minimum duration = F ÷ overhead target (1 s), at least 10 tasks per worker
Is speculation safe?	Utilization, determinism, detector	Idle capacity, deterministic tasks, remaining-time estimate
How wide can a shuffle be?	b against b_min	M = R ≤ √(S/b_min); reduce S before changing topology
Which shuffle format?	R and consumer needs	Small R: hash bypass. Large R: sort. Sorted consumer: sort by key on the map side
Compress shuffle data?	Link speed and codec speed	Only if C > BW × r/(r−1)
Retries and speculation allowed?	Partition function	Verify determinism first

Patterns that repeat:

Fixed cost per operation gives a natural unit size: batch size, compile break-even, morsel size, task size, shuffle fragment size.
Granularity mismatch amplifies work, and granularity is the floor on adaptivity: no mechanism corrects below the unit it schedules. Skew splitting even depended on a file format choice.
Costs are moved, not removed. Hash shuffle bills memory, sort shuffle bills a cheap counting pass, push-based shuffle bills a third party and a second hop.
Variance costs more than volume. Barriers turn variance into latency.
Queues explode near full load, which is why speculation turns negative when the cluster is busy.
Uniformity is assumed and false. Partitioners assume even keys; Zipf turns that into a closed-form ceiling.
Randomization can replace coordination (work stealing, with a proof).
Algebra gates topology. Only associative reductions can be combined in parts, in trees or before a shuffle.
Immutable output grows an indirection layer plus a garbage collector (the map-output tracker and the cleaner).
Control flow is flow control. Whoever decides when the next unit moves is the only one who can refuse it: pull gets backpressure free, push must build it.

Key formulas and numbers:

Quantity	Value
Work-span bound	T_p ≥ max(T₁/p, T∞)
Data-parallel ceiling	1/f; Zipf-1: 1/H_n, H_n ≈ ln n + 0.577 (about 14 at 10⁶ keys, about 21 at 10⁹)
Minimum task duration	F ÷ overhead target: 10 ms ÷ 1% = 1 s
Elasticity ceiling	total_work / (10 × min_task) workers
Bytes per connection	b = S / (M × R)
Width ceiling	M = R ≤ √(S/b_min): 100 GB about 316; 1 TB 1000; 10 TB about 3162
Spill amplification	1 + ⌈log_k(S_task/B)⌉: three passes at ratio 1000 and k = 64
Hash shuffle buffers	cores × R × buffer: 16 × 1000 × 32 KB = 512 MB
Broadcast threshold	(R − 1) × S_small < S_large
Compression rule	C > BW × r/(r−1)
Object-store request cost	$5 / (10⁶ × object size in GB): $0.05 per GB at 100 KB
Streaming vs materializing	n × T × λ against 1

What the runtime cannot do. It cannot know how long a piece of work will take before it runs. It cannot correct an imbalance smaller than the unit it schedules, or split a unit that is too big by moving it. It cannot parallelize an operator whose algebra is not decomposable. It cannot make a transpose cheap: the shuffle only chooses who pays. Every escape has a price, paid in memory, duplicate work, a second network hop, a stateful service or lost determinism.