14. Big Data and Distributed Computing

Every technique in this guide has assumed the data fits somewhere you can reach it -- a dataframe in memory, a table a query returns in seconds. This chapter covers what changes when it does not: how work is split across machines that fail independently, why moving bytes becomes the dominant cost, how unbounded streams differ from finite tables, and how to find and fix the bottleneck in a job that takes six hours. The first lesson is deflationary: most datasets called "big" are not, and a cluster added too early makes everything slower and more expensive.

<a id="141-scaling-concepts"></a>

14.1 Scaling Concepts

Distributed systems are built from a small number of ideas that recur in every framework you will meet; this section covers the choice between a bigger machine and more machines, how files are stored across nodes that fail, the map-shuffle-reduce shape that almost all batch processing still follows, why data movement rather than computation sets the runtime, and what you are forced to give up once state lives in more than one place.

Vertical vs Horizontal Scaling

There are exactly two ways to make a computation handle more data. Vertical scaling (scaling up) replaces the machine with a larger one: more cores, more RAM, faster disks. Horizontal scaling (scaling out) keeps the machines and adds more of them, splitting the work.

Vertical scaling is enormously underrated. A single cloud instance can be rented today with 128 cores and 2 TB of RAM. Inside it, your code keeps every convenience it had on a laptop: shared memory, no serialisation, no network, no partial failure, a debugger that works, and a stack trace that means something. The moment you scale out, you inherit a distributed system, and with it network latency, serialisation cost, stragglers, partial failures, and a class of bug that only appears at 3 a.m. on 200 nodes.

So why does anyone scale out? Three reasons, in order of how often they actually apply:

1. The data genuinely exceeds the largest machine. Petabyte-scale logs, not gigabyte-scale tables.
2. Cost curves bend. Machine price grows faster than linearly in RAM at the top of the range, so 16 medium nodes can be cheaper than one large one for the same total memory.
3. Fault tolerance is a requirement. One machine is one failure domain. A pipeline that must complete tonight cannot be hostage to a single host's disk.

The limit on scaling out is not engineering effort -- it is arithmetic. Amdahl's law says that if a fraction p of the work is parallelisable, the speedup with n workers is 1 / ((1 - p) + p/n). A job that is 95% parallel can never go more than 20x faster, no matter how many machines you rent. Worse, real systems have a term Amdahl's law omits: workers must coordinate with each other, and that cost grows with the square of the worker count. The Universal Scalability Law adds it, and produces the curve every practitioner eventually observes -- throughput rises, flattens, and then falls.

  THROUGHPUT VS WORKERS, THE SHAPE YOU ACTUALLY GET

    x14 |                  ****
        |              ****    ****
    x10 |          ****            ****
        |       ***                    *****
     x6 |    ***                            ******
        |  **                                     *****
     x2 | *
        +----+----+----+----+----+----+----+----+----+---
          1    4    8   16   32   64  128  256  512  workers
                          ^ peak
        Left of the peak: adding workers helps.
        Right of the peak: workers spend their time talking to each other.

def amdahl(n, p):
    """Speedup with a fraction p of the work parallelisable."""
    return 1.0 / ((1.0 - p) + p / n)

def usl(n, sigma, kappa):
    """Universal Scalability Law: contention (sigma) plus coherency (kappa)."""
    return n / (1 + sigma * (n - 1) + kappa * n * (n - 1))

print("AMDAHL: SPEEDUP CEILING FOR A JOB THAT IS 95% PARALLELISABLE")
for n in (1, 2, 4, 8, 16, 64, 256, 1024):
    print(f"  {n:5d} workers -> speedup {amdahl(n, 0.95):7.2f}"
          f"   efficiency {amdahl(n, 0.95) / n * 100:5.1f}%")
print(f"  infinite workers -> speedup {1 / 0.05:7.2f}  (the 5% serial part)")

print("\nUSL: REAL SYSTEMS GET SLOWER PAST A POINT")
best_n, best_s = 0, 0.0
for n in (1, 2, 4, 8, 16, 32, 64, 128, 256):
    s = usl(n, sigma=0.03, kappa=0.0005)     # 3% contention, small coherency
    if s > best_s:
        best_n, best_s = n, s
    print(f"  {n:5d} workers -> throughput x{s:6.2f}")
print(f"  peak at {best_n} workers (x{best_s:.2f}); beyond that,")
print("  coordination costs more than the added work")

GB = 1024 ** 3
print("\nDOES THE DATA ACTUALLY NEED A CLUSTER?")
for name, rows, bytes_per_row in [
    ("daily event extract",       50_000_000,  120),
    ("one year of orders",       900_000_000,   64),
    ("raw clickstream, 1 year", 40_000_000_000, 250),
]:
    raw = rows * bytes_per_row / GB
    # Columnar encoding plus compression typically lands near 5-6x on
    # event data with low-cardinality string columns.
    print(f"  {name:24s} {raw:9.1f} GB raw"
          f"  {raw * 0.18:8.1f} GB columnar+compressed")
print("  A 512 GB machine handles the first two in memory. Only the third")
print("  has no single-machine answer.")

  AMDAHL: SPEEDUP CEILING FOR A JOB THAT IS 95% PARALLELISABLE
        1 workers -> speedup    1.00   efficiency 100.0%
        2 workers -> speedup    1.90   efficiency  95.2%
        4 workers -> speedup    3.48   efficiency  87.0%
        8 workers -> speedup    5.93   efficiency  74.1%
       16 workers -> speedup    9.14   efficiency  57.1%
       64 workers -> speedup   15.42   efficiency  24.1%
      256 workers -> speedup   18.62   efficiency   7.3%
     1024 workers -> speedup   19.64   efficiency   1.9%
    infinite workers -> speedup   20.00  (the 5% serial part)

  USL: REAL SYSTEMS GET SLOWER PAST A POINT
        1 workers -> throughput x  1.00
        2 workers -> throughput x  1.94
        4 workers -> throughput x  3.65
        8 workers -> throughput x  6.46
       16 workers -> throughput x 10.19
       32 workers -> throughput x 13.19
       64 workers -> throughput x 13.05
      128 workers -> throughput x  9.89
      256 workers -> throughput x  6.20
    peak at 32 workers (x13.19); beyond that,
    coordination costs more than the added work

  DOES THE DATA ACTUALLY NEED A CLUSTER?
    daily event extract            5.6 GB raw       1.0 GB columnar+compressed
    one year of orders            53.6 GB raw       9.7 GB columnar+compressed
    raw clickstream, 1 year     9313.2 GB raw    1676.4 GB columnar+compressed
    A 512 GB machine handles the first two in memory. Only the third
    has no single-machine answer.

The 5.6 GB row is the one worth staring at. Fifty million rows sounds like "big data" in a planning meeting, and it fits comfortably in a laptop's RAM once stored columnar. A Spark cluster processing it will very likely be slower than a single-process library, because it pays for JVM startup, task scheduling, serialisation, and shuffle on a dataset that never needed any of them.

	Vertical (one big machine)	Horizontal (many machines)
Programming model	Ordinary single-process code	Partitioned, failure-aware
Debugging	Normal tooling, real stack traces	Logs across N hosts
Failure domain	One; the job dies with the host	Task-level retry
Data ceiling	A few TB in RAM today	Effectively unbounded
Latency floor	Memory bandwidth	Network round trips
Cost at small scale	Lower	Higher (overhead dominates)
Cost at large scale	Superlinear, then impossible	Roughly linear

The practical rule: scale up until you cannot, then scale out. Before provisioning a cluster, load the data as Parquet, keep only the columns the job needs, and try it on one machine with a library designed for that case. Many "we need Spark" conclusions dissolve at that step.

Key Takeaways

• Vertical scaling buys a bigger machine; horizontal scaling adds machines and hands you a distributed system along with them.
• A single modern instance offers ~100 cores and terabytes of RAM, which covers far more workloads than most teams assume.
• Amdahl's law caps speedup at 1/(1-p): a 95% parallel job cannot exceed 20x however many workers you add.
• Real throughput follows the Universal Scalability Law -- it peaks and then declines, because coordination cost grows quadratically.
• Estimate compressed columnar size, not row count, when deciding whether a cluster is needed.
• Scale out for data that exceeds one machine, for bending cost curves, or for fault tolerance -- not because the row count has many digits.
• Scale up first; measure; scale out only when the single machine is genuinely the binding constraint.

🧪 Practice

1. Compute Amdahl speedup for p = 0.9, 0.99, 0.999 at 8, 64, and 1024 workers, and state the ceiling for each.
2. Fit the Universal Scalability Law to a throughput measurement at 1, 2, 4, 8, and 16 workers, then predict where the peak is.
3. Take a CSV you have, convert it to Parquet keeping only the columns a specific query needs, and report the size ratio.
4. For a workload of 2 billion rows at 80 bytes per row, work out whether a 768 GB single instance or a 24-node cluster is cheaper using your cloud provider's current on-demand prices.
5. Interview: A team reports that their Spark job on 40 executors is slower than the Pandas script it replaced. Give three plausible causes. (Hint: think about what the framework must do before any of your code runs, and what fraction of a small dataset each executor ends up with.)

Distributed File Systems

Once data spans machines, the first thing that must be rebuilt is the file. A single-machine filesystem assumes one disk controller, one directory tree, and that a write either happens or the machine is off. None of that holds across a hundred hosts.

A distributed file system solves this by splitting every file into large fixed-size blocks and replicating each block onto several machines. HDFS made the design famous; cloud object stores such as S3, GCS, and Azure Blob apply the same principles behind a different API. Two roles exist:

• A metadata service (the HDFS NameNode) knows the directory tree and, for each file, the ordered list of blocks and which nodes hold each copy. It holds no file data.
• Data nodes store opaque blocks and report in with heartbeats. They know nothing about filenames.

  ONE 640 MB FILE, 128 MB BLOCKS, REPLICATION FACTOR 3

    METADATA SERVICE
      /events/2026-09-28.parquet
        block b1 -> [n1, n5, n9]      no file bytes stored here,
        block b2 -> [n2, n6, n10]     only the map from blocks to nodes
        block b3 -> [n3, n7, n11]
        block b4 -> [n1, n8, n12]
        block b5 -> [n4, n5, n10]

    RACK A            RACK B            RACK C
    n1 [b1 b4]        n5 [b1 b5]        n9  [b1]
    n2 [b2]           n6 [b2]           n10 [b2 b5]
    n3 [b3]           n7 [b3]           n11 [b3]
    n4 [b5]           n8 [b4]           n12 [b4]

    A reader asks the metadata service where b3 is, then streams it
    directly from whichever of n3, n7, n11 is closest. The metadata
    service never touches the data path.

Three design decisions explain most of the behaviour you will observe.

Blocks are huge. 128 MB is typical, against 4 KB on a local filesystem. The reason is the seek-to-transfer ratio: a disk seek costs ~10 ms while transferring 128 MB costs ~1 s, so a large block spends 99% of its time doing useful sequential I/O. The consequence is the notorious small files problem: a million 10 KB files consume a million metadata entries (all held in the metadata service's memory) and spawn a million tasks that each do almost no work. Compaction into few large files is often the single highest -leverage fix available to a slow pipeline.

Writes are append-only. You can create a file and append to it; you cannot overwrite byte 4,000,000 in place. This sounds restrictive until you notice that it removes an entire class of distributed coordination problem, and that analytics never needs in-place updates anyway -- it needs new partitions. Table formats such as Delta Lake, Iceberg, and Hudi build updates, deletes, and transactions on top of immutable files, by writing new files plus a metadata log that says which files are currently live.

Replication is placement-aware. Three copies of a block on three machines in the same rack survive three disk failures but not one rack's top-of-rack switch. The standard policy places the first replica on the writer's node, the second on a different rack, and the third on that same second rack -- buying rack-failure survival without paying for three cross-rack transfers.

The modern variant separates storage from compute entirely. Object storage holds the data; clusters are created for a query and destroyed after it. You lose data locality (every read crosses the network) and gain elasticity, and the trade has proven worthwhile because network bandwidth inside a datacentre grew far faster than disk throughput.

import os, random, tempfile
import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.csv as pv

rng = random.Random(0)
n = 2_000_000
tbl = pa.table({
    "event_ts": pa.array([1_700_000_000 + i for i in range(n)], pa.int64()),
    "user_id":  pa.array([rng.randrange(50_000) for _ in range(n)], pa.int32()),
    "country":  pa.array([rng.choice(["IT", "DE", "FR", "ES", "PL"])
                          for _ in range(n)]),
    "amount":   pa.array([round(rng.expovariate(1 / 40), 2) for _ in range(n)]),
})

d = tempfile.mkdtemp()
csv_p, pq_p = os.path.join(d, "e.csv"), os.path.join(d, "e.parquet")
pv.write_csv(tbl, csv_p)
pq.write_table(tbl, pq_p, compression="zstd")      # columnar + compressed

MB = 1024 ** 2
print("SAME 2,000,000 ROWS, TWO STORAGE LAYOUTS")
print(f"  row-oriented CSV        {os.path.getsize(csv_p) / MB:8.1f} MB")
print(f"  columnar Parquet+zstd   {os.path.getsize(pq_p) / MB:8.1f} MB")

# A columnar layout lets a reader fetch ONE column's byte ranges and skip
# the rest of the file entirely -- the property that makes cloud object
# storage viable for analytics.
one = pq.read_table(pq_p, columns=["amount"])
md = pq.ParquetFile(pq_p).metadata
col_bytes = sum(md.row_group(g).column(3).total_compressed_size
                for g in range(md.num_row_groups))
print(f"  bytes read for SELECT amount: {col_bytes / MB:.1f} MB"
      f"  ({col_bytes / os.path.getsize(pq_p) * 100:.0f}% of the file)")
print(f"  rows returned: {one.num_rows:,}")

# --- REPLICATION: WHY THREE COPIES, AND WHY THEIR PLACEMENT MATTERS. ---
def block_survives(replicas, node_fail_p, rack_fail_p, racks_of):
    """A block is readable if at least one replica sits on a live node in a
    rack whose switch is up."""
    # sorted() keeps the draw order stable across processes
    dead_racks = {r for r in sorted(set(racks_of.values()))
                  if random.random() < rack_fail_p}
    return any(racks_of[nd] not in dead_racks and random.random() > node_fail_p
               for nd in replicas)

random.seed(7)
racks_of = {f"n{i}": f"rack{i // 4}" for i in range(12)}   # 12 nodes, 3 racks
naive = ["n0", "n1", "n2"]          # all three copies in rack0
aware = ["n0", "n4", "n5"]          # one in rack0, two in rack1
TRIALS = 200_000
for name, repl in [("all replicas in one rack", naive),
                   ("rack-aware placement", aware)]:
    lost = sum(not block_survives(repl, 0.01, 0.005, racks_of)
               for _ in range(TRIALS))
    print(f"\n  {name:26s} replicas={repl}")
    print(f"    block-loss rate {lost / TRIALS * 100:.3f}%"
          f"  over {TRIALS:,} simulated days")

  SAME 2,000,000 ROWS, TWO STORAGE LAYOUTS
    row-oriented CSV            52.5 MB
    columnar Parquet+zstd       10.8 MB
    bytes read for SELECT amount: 3.7 MB  (34% of the file)
    rows returned: 2,000,000

    all replicas in one rack   replicas=['n0', 'n1', 'n2']
      block-loss rate 0.507%  over 200,000 simulated days

    rack-aware placement       replicas=['n0', 'n4', 'n5']
      block-loss rate 0.009%  over 200,000 simulated days

Identical replica count, identical hardware, a 57x difference in data loss rate -- produced entirely by where the copies were put. This is the general shape of distributed storage decisions: the redundancy level is the obvious knob, and the failure-domain layout is the one that actually determines durability.

Key Takeaways

• A distributed file system splits files into large blocks and replicates each block across machines; metadata and data are handled by separate services.
• Blocks are ~128 MB so that sequential transfer dominates seek time; millions of small files break both the metadata service and the task scheduler.
• Files are immutable and append-only; updates and deletes come from table formats (Delta, Iceberg, Hudi) layering a metadata log over new files.
• Replica placement across failure domains matters more than replica count -- three copies in one rack is close to one copy.
• Columnar formats let a reader fetch only the columns a query needs, which is what makes remote object storage practical for analytics.
• Separating storage from compute trades data locality for elasticity, and fast datacentre networks have made that trade a good one.

🧪 Practice

1. Write 500 tiny Parquet files and one large Parquet file with the same total rows, then compare the time to read each set.
2. Compute how many blocks a 12 GB file occupies at block sizes of 64 MB, 128 MB, and 512 MB, and how many map tasks each implies.
3. Extend the replication simulation to replication factor 2 and 4, and plot block-loss rate against factor for both placement policies.
4. Using a Parquet file's metadata, report per-column compressed sizes and identify which column a projection should avoid reading.
5. Interview: Your pipeline writes 40,000 files per day into object storage and downstream queries have got slower every week. Explain the mechanism and the fix. (Hint: think about what a query planner must do before reading any row, and what a compaction job would change.)

MapReduce Paradigm

MapReduce is a way of describing a computation so that a framework -- not you -- can decide how to parallelise it, where to run each piece, and what to do when a machine dies mid-job. It has been superseded as an engine, but its shape survives in Spark, Flink, BigQuery, and every distributed SQL planner, so it is the right thing to learn first.

The insight is that a huge share of data processing fits one template:

1. Map: apply a function to each record independently, emitting zero or more (key, value) pairs. Independence is the point -- any record can be processed on any machine, in any order.
2. Shuffle: group all pairs with the same key onto the same machine. This step is not written by the user, and it is where the time goes.
3. Reduce: for each key, combine its values into a result.

Because map tasks are independent and reduce tasks depend only on their own key group, the framework gets fault tolerance almost free: if a task dies, re-run that task. Nothing else is affected, because nothing shared was mutated.

  WORD COUNT, THREE PHASES

  INPUT SPLITS        MAP                SHUFFLE              REDUCE
  ------------        ---                -------              ------
  "the data node" -> (the,1)(data,1) --\
                     (node,1)           \   partition 0
  "the the node"  -> (the,1)(the,1) -----+> (data,[1,1])  -> data  2
                     (node,1)           /   (node,[1,1,1])-> node  3
  "data node key" -> (data,1)(node,1)--/
                     (key,1)            \   partition 1
                                         +> (the,[1,1,1])  -> the   3
                                            (key,[1])      -> key   1

  Map is embarrassingly parallel. Reduce is parallel across keys.
  The arrows in the middle are network traffic, and they are the bill.

The dominant cost is the shuffle, because it is the only phase that moves data between machines. This is why MapReduce has a fourth, optional component that every practitioner should know: the combiner. A combiner is a reduce-like function applied locally on the map output, before the shuffle. For word count, combining (the,1)(the,1)(the,1) into (the,3) on the mapper cuts network traffic by orders of magnitude.

A combiner is only legal when the reduce operation is associative and commutative over partial results, because the framework may apply it zero, one, or many times, in any grouping. Sum, count, min, max qualify. Mean, median, and distinct-count do not -- but mean can be made to qualify by emitting (sum, count) pairs instead of values, which is a trick worth internalising: when an aggregation resists distribution, look for a richer intermediate value that is associative.

import random
from collections import defaultdict

# --- A 25-LINE MAPREDUCE ENGINE. THE REAL ONES ADD FAULT TOLERANCE AND
# --- NETWORK TRANSFER; THE CONTROL FLOW IS EXACTLY THIS.
def map_reduce(splits, mapper, reducer, n_reducers=3, combiner=None):
    stats = {"map_out": 0, "shuffle_out": 0}
    buckets = defaultdict(list)                   # what crosses the network
    for split in splits:                          # one map task per split
        emitted = [kv for rec in split for kv in mapper(rec)]
        stats["map_out"] += len(emitted)
        if combiner:                              # pre-aggregate LOCALLY
            local = defaultdict(list)
            for k, v in emitted:
                local[k].append(v)
            emitted = [(k, combiner(k, vs)) for k, vs in local.items()]
        stats["shuffle_out"] += len(emitted)
        for k, v in emitted:
            buckets[hash(k) % n_reducers].append((k, v))   # partition by key
    out = {}
    for _, pairs in sorted(buckets.items()):      # one reduce task per bucket
        grouped = defaultdict(list)
        for k, v in sorted(pairs, key=lambda p: p[0]):     # the "sort" phase
            grouped[k].append(v)
        for k, vs in grouped.items():
            out[k] = reducer(k, vs)
    return out, stats

WORDS = ["the", "and", "data", "cluster", "shuffle", "node", "key", "spill"]
rng = random.Random(1)
weights = [0.40, 0.22, 0.14, 0.09, 0.06, 0.05, 0.03, 0.01]   # Zipf-like
lines = [" ".join(rng.choices(WORDS, weights, k=12)) for _ in range(6000)]
splits = [lines[i:i + 1000] for i in range(0, len(lines), 1000)]   # 6 splits

def mapper(line):   return [(w, 1) for w in line.split()]
def reducer(k, vs): return sum(vs)
def combiner(k, vs): return sum(vs)

plain, s1 = map_reduce(splits, mapper, reducer)
combo, s2 = map_reduce(splits, mapper, reducer, combiner=combiner)

print("WORD COUNT OVER 6 SPLITS, 72,000 WORDS")
for w in WORDS:
    print(f"  {w:8s} {plain[w]:7,}")
assert plain == combo, "a correct combiner cannot change the answer"
print("  combiner produces identical counts:", plain == combo)

print("\nWHAT CROSSES THE NETWORK")
print(f"  no combiner : map emitted {s1['map_out']:,}"
      f" -> shuffled {s1['shuffle_out']:,} pairs")
print(f"  combiner    : map emitted {s2['map_out']:,}"
      f" -> shuffled {s2['shuffle_out']:,} pairs")
print(f"  reduction   : {100 * (1 - s2['shuffle_out'] / s1['shuffle_out']):.1f}%"
      f" fewer pairs over the wire")

# --- WHEN A COMBINER IS ILLEGAL: THE REDUCE MUST BE ASSOCIATIVE AND
# --- COMMUTATIVE OVER PARTIAL RESULTS.
sales = ([("IT", round(rng.expovariate(1 / 30), 2)) for _ in range(9000)]
         + [("IT", round(rng.expovariate(1 / 300), 2)) for _ in range(200)])
# Deliberately uneven splits: the small one holds the high-value orders,
# which is what partitioning by date or region tends to produce.
uneven = [sales[0:3000], sales[3000:6000], sales[6000:9000], sales[9000:]]

amount   = lambda rec: [(rec[0], rec[1])]
mean     = lambda k, vs: sum(vs) / len(vs)
pair     = lambda rec: [(rec[0], (rec[1], 1))]      # emit (sum, count)
pair_agg = lambda k, vs: (sum(a for a, _ in vs), sum(c for _, c in vs))
pair_div = lambda k, vs: (lambda t: t[0] / t[1])(pair_agg(k, vs))

truth, _  = map_reduce(uneven, amount, mean)                  # no combiner
broken, _ = map_reduce(uneven, amount, mean, combiner=mean)   # mean of means
fixed, _  = map_reduce(uneven, pair, pair_div, combiner=pair_agg)

print(f"\nMEAN ORDER VALUE, SPLIT SIZES {[len(s) for s in uneven]}")
print(f"  no combiner (correct)     {truth['IT']:8.2f}")
print(f"  mean-of-means combiner    {broken['IT']:8.2f}"
      f"   <-- {broken['IT'] / truth['IT'] - 1:.0%} too high")
print(f"  (sum, count) combiner     {fixed['IT']:8.2f}")
print("  A mean is not associative: averaging four split means weights a")
print("  200-row split the same as a 3,000-row one. Emit (sum, count) and")
print("  the operation becomes associative, so the combiner is safe.")

  WORD COUNT OVER 6 SPLITS, 72,000 WORDS
    the       28,638
    and       16,011
    data       9,930
    cluster    6,399
    shuffle    4,439
    node       3,576
    key        2,265
    spill        742
    combiner produces identical counts: True

  WHAT CROSSES THE NETWORK
    no combiner : map emitted 72,000 -> shuffled 72,000 pairs
    combiner    : map emitted 72,000 -> shuffled 48 pairs
    reduction   : 99.9% fewer pairs over the wire

  MEAN ORDER VALUE, SPLIT SIZES [3000, 3000, 3000, 200]
    no combiner (correct)        35.54
    mean-of-means combiner       95.77   <-- 169% too high
    (sum, count) combiner        35.54
    A mean is not associative: averaging four split means weights a
    200-row split the same as a 3,000-row one. Emit (sum, count) and
    the operation becomes associative, so the combiner is safe.

Two results to carry forward. The combiner cut shuffled pairs from 72,000 to 48 without changing a single count -- that is the entire argument for pre-aggregation. And the illegal combiner produced an answer that was 169% wrong while raising no error at all, which is the characteristic failure mode of distributed aggregation: silently plausible numbers.

Why did MapReduce get replaced? Because every stage boundary wrote to disk. A five-step pipeline meant five full round trips through the distributed filesystem, which is catastrophic for iterative algorithms -- and gradient descent, k-means, and PageRank are all iterative. Spark's central idea was to keep intermediate results in memory between stages while preserving the same fault-tolerance story.

Key Takeaways

• MapReduce expresses a computation as independent per-record map, a framework-managed shuffle that groups by key, and a per-key reduce.
• Independence is what buys parallelism and fault tolerance: a failed task is simply re-run.
• The shuffle is the only phase that moves data between machines, so it dominates cost and is the first thing to attack.
• Combiners pre-aggregate on the mapper and can cut shuffle volume by orders of magnitude.
• A combiner is legal only for associative, commutative reductions; a mean-of-means combiner gives wrong answers silently.
• Make a stubborn aggregation distributable by choosing a richer associative intermediate, such as (sum, count) for a mean.
• MapReduce's disk-per-stage design is why in-memory engines replaced it, especially for iterative algorithms.

🧪 Practice

1. Implement word count with the engine above on a text file of your own and confirm the combiner leaves the counts unchanged.
2. Write a MapReduce formulation of "average session length per country" using an associative intermediate value.
3. Express an inverted index (term -> list of document ids) as a map-reduce pair, and say what the reduce output size depends on.
4. Implement a distributed distinct-count as map-reduce, first exactly, then approximately with a HyperLogLog sketch, and compare shuffle volume.
5. Interview: Which of sum, max, mean, median, and count distinct can use a combiner as written, and how would you rewrite the ones that cannot? (Hint: ask whether combining two partial results can be done with only those partial results in hand.)

Data Locality and Shuffling

Two numbers explain almost every distributed performance problem. Reading a gigabyte from local memory takes on the order of a tenth of a second; moving a gigabyte across a datacentre network takes several seconds; moving it over a saturated link takes much longer. Computation is cheap and plentiful. Data movement is neither.

Data locality is the strategy that follows: send the computation to the machine that already holds the data, rather than the data to the machine running the code. The map phase of a well-scheduled job reads almost entirely from local disk. Schedulers express this as a preference ladder, and will wait briefly for a better rung before settling:

Level	Meaning	Relative cost
PROCESS_LOCAL	Data already in this executor's memory	1x
NODE_LOCAL	On this machine's disk	~2-5x
RACK_LOCAL	Same rack, one switch hop	~10x
ANY	Anywhere in the cluster	~20x+

Locality is achievable for map-like work. It is impossible for anything that groups by key, because the rows sharing a key start out scattered by construction. That operation is the shuffle, and understanding its mechanics is the difference between tuning a job and guessing at it.

  WHAT A SHUFFLE ACTUALLY DOES

  STAGE 1 (map side)                      STAGE 2 (reduce side)
  each task partitions its output         each task pulls its own
  by hash(key) % numPartitions,           partition file from every
  sorts it, spills it to local disk       upstream task, then merges

    task A  [p0|p1|p2|p3]  --\ \ \
    task B  [p0|p1|p2|p3]  ---+-\-\--->  task for p0 reads A.p0, B.p0,
    task C  [p0|p1|p2|p3]  ----+-+---->  C.p0, D.p0 ... and merges them
    task D  [p0|p1|p2|p3]  -----+/ /

  M map tasks x R reduce tasks = M*R transfer requests.
  At M = R = 1,000 that is a million fetches, which is why raising
  parallelism eventually makes a shuffle slower, not faster.

The failure mode that costs the most engineering time is skew. Hash partitioning distributes keys evenly only if the keys themselves are distributed evenly, and real keys never are: one customer is a marketplace reselling for thousands, one user_id is a load-test account, and NULL outnumbers every real value. A stage finishes when its slowest task finishes, so a single hot partition sets the wall clock for the whole stage, while 200 executors sit idle.

The standard remedies, in the order you should reach for them:

1. Broadcast the small side of a join. If one input fits in an executor's memory, copy it everywhere and skip the shuffle entirely.
2. Filter and project before shuffling. Every column and row dropped early is bytes not serialised, not written, not fetched, not merged.
3. Pre-aggregate. reduceByKey over groupByKey; partial aggregation before the exchange.
4. Salt the hot key. Append a random suffix so the hot key spreads over many partitions, then re-aggregate the partials in a cheap second stage.
5. Isolate the hot keys. Process the top few keys in a separate job with a different strategy, and union the results.

import random
from collections import Counter
from zlib import crc32                  # stable across processes, unlike hash()

rng = random.Random(11)
N_PART = 8

# --- A REALISTIC KEY DISTRIBUTION: ONE CUSTOMER IS A MARKETPLACE. ---
keys = ([f"cust{rng.randrange(5000)}" for _ in range(400_000)]
        + ["cust0"] * 600_000)          # cust0 alone is 60% of the rows
rng.shuffle(keys)

def partition_of(key, n):
    return crc32(key.encode()) % n      # exactly what a hash partitioner does

def sizes_of(keys, n, key_fn=lambda k: k):
    c = Counter(partition_of(key_fn(k), n) for k in keys)
    return [c.get(i, 0) for i in range(n)]

def report(label, sizes):
    total, mx, mean = sum(sizes), max(sizes), sum(sizes) / len(sizes)
    print(f"  {label}")
    print("    rows per partition  "
          + " ".join(f"{s // 1000:4d}k" for s in sizes))
    print(f"    largest partition   {mx / total:.0%} of all rows,"
          f" {mx / mean:.1f}x the mean")
    # A stage ends when its SLOWEST task ends, not its average one.
    print(f"    stage wall clock    ~{mx / mean:.1f}x a balanced stage")

print(f"HASH PARTITIONING {len(keys):,} ROWS ONTO {N_PART} PARTITIONS")
report("PLAIN KEY", sizes_of(keys, N_PART))

# --- SALTING: SPLIT THE HOT KEY, THEN RE-AGGREGATE IN A SECOND STAGE. ---
SALTS = 64
salted = [(k, rng.randrange(SALTS)) for k in keys]
print(f"\nAFTER SALTING EVERY KEY WITH A RANDOM 0..{SALTS - 1} SUFFIX")
report("SALTED KEY", sizes_of(salted, N_PART,
                              key_fn=lambda t: f"{t[0]}#{t[1]}"))
print("  Stage two re-aggregates the 64 partials per key: one extra shuffle")
print("  of at most 64 rows per key, in exchange for no hot partition.")

# --- BROADCAST JOIN VS SHUFFLE JOIN: COUNT THE BYTES MOVED. ---
MB = 1024 ** 2
fact_rows, fact_row_bytes = 800_000_000, 90       # orders
dim_rows, dim_row_bytes = 5_000, 200              # customers
workers = 40

shuffle_bytes = fact_rows * fact_row_bytes + dim_rows * dim_row_bytes
broadcast_bytes = dim_rows * dim_row_bytes * workers

print(f"\nJOINING 800M ORDERS TO 5,000 CUSTOMERS ON {workers} WORKERS")
print(f"  shuffle join    {shuffle_bytes / MB:>9,.0f} MB moved"
      "   both sides re-partitioned")
print(f"  broadcast join  {broadcast_bytes / MB:>9,.1f} MB moved"
      "   small side copied per worker")
print(f"  broadcast moves {shuffle_bytes / broadcast_bytes:,.0f}x less data")
print("  It stops being viable once the small side no longer fits in one")
print("  executor's memory -- tens to low hundreds of MB in practice.")

  HASH PARTITIONING 1,000,000 ROWS ONTO 8 PARTITIONS
    PLAIN KEY
      rows per partition    50k  649k   49k   50k   50k   50k   49k   49k
      largest partition   65% of all rows, 5.2x the mean
      stage wall clock    ~5.2x a balanced stage

  AFTER SALTING EVERY KEY WITH A RANDOM 0..63 SUFFIX
    SALTED KEY
      rows per partition   115k  125k  134k  124k  134k  124k  115k  125k
      largest partition   13% of all rows, 1.1x the mean
      stage wall clock    ~1.1x a balanced stage
    Stage two re-aggregates the 64 partials per key: one extra shuffle
    of at most 64 rows per key, in exchange for no hot partition.

  JOINING 800M ORDERS TO 5,000 CUSTOMERS ON 40 WORKERS
    shuffle join       68,666 MB moved   both sides re-partitioned
    broadcast join       38.1 MB moved   small side copied per worker
    broadcast moves 1,800x less data
    It stops being viable once the small side no longer fits in one
    executor's memory -- tens to low hundreds of MB in practice.

Salting turned a 5.2x straggler into a 1.1x one, and broadcasting replaced 68 GB of network traffic with 38 MB. Neither change touched the business logic. This is characteristic: in distributed processing the large wins come from changing how data is placed and moved, not from optimising the function applied to each row.

Key Takeaways

• Computation is cheap and data movement is expensive, so schedulers send code to data rather than the reverse.
• Locality is achievable for map-style work and impossible for anything that groups by key -- that is what a shuffle is.
• A shuffle writes partitioned, sorted files on the map side and fetches them on the reduce side; M x R tasks means M*R fetches.
• Skew is the dominant shuffle pathology: a stage takes as long as its slowest task, so one hot key idles the whole cluster.
• Fix skew by broadcasting small sides, filtering early, pre-aggregating, salting hot keys, or isolating them into a separate job.
• Broadcast joins eliminate the shuffle entirely when one side fits in executor memory, often a 1000x reduction in bytes moved.
• More parallelism is not monotonically better: past a point the fetch count itself becomes the bottleneck.

🧪 Practice

1. Build a key distribution where one key holds 70% of rows, hash it onto 16 partitions, and report the skew factor.
2. Apply salting with 8, 32, and 128 salts to that distribution and plot the largest partition's share against salt count.
3. For a 2 TB fact table joined to a 40 MB dimension table across 60 executors, compute bytes moved under shuffle and broadcast joins.
4. Take any groupByKey-style aggregation you have written and rewrite it as a pre-aggregating reduction; measure the difference in shuffled bytes if your engine reports them.
5. Interview: One task in a 400-task stage has been running for 50 minutes while the other 399 finished in 30 seconds. Walk through your diagnosis. (Hint: start from what determines which rows that one task received, and check the null handling in the join key.)

CAP Theorem and Consistency Models

The moment state is replicated across machines, a guarantee that was free on one machine stops being free: that everyone reading the data sees the same thing. The CAP theorem states the constraint precisely. A distributed system can offer at most two of:

• Consistency: every read returns the most recent committed write.
• Availability: every request receives a non-error response.
• Partition tolerance: the system keeps working when the network drops messages between nodes.

The framing invites a mistake. Partitions are not a design choice -- networks fail, switches reboot, cables get unplugged -- so P is mandatory, and the theorem reduces to a single question: when a partition happens, do you refuse requests (CP) or serve possibly stale data (AP)?

  A 5-NODE CLUSTER, NETWORK SPLIT 3 | 2

        MAJORITY SIDE            |            MINORITY SIDE
        n1   n2   n3             X            n4   n5
                                 |
    CP  reads OK, writes OK      |     refuses: "no quorum"
        one truth preserved      |     clients there see an outage

    AP  reads OK, writes OK      |     reads OK, writes OK
        both sides accept writes and diverge; when the partition
        heals you must reconcile -- last-write-wins, vector clocks,
        or a CRDT whose merge is defined to be conflict-free

CAP also under-describes normal operation, because partitions are rare and the interesting trade-off happens the other 99.9% of the time. PACELC extends it: if there is a Partition, choose Availability or Consistency; Else, choose Latency or Consistency. Strong consistency costs a network round trip on every read even when nothing is broken, and that latency is what most systems are actually trading away.

Between "always correct" and "eventually, probably" sits a spectrum:

Model	Guarantee	Typical cost
Linearizable	Reads see the latest write, globally ordered	Consensus round trip per operation
Sequential	All nodes see operations in the same order	Ordered log, no real-time bound
Causal	Related operations are seen in order; unrelated may differ	Track dependencies
Read-your-writes	A client sees its own writes	Pin to replica, or read from primary
Monotonic reads	Never see time go backwards	Sticky sessions
Eventual	Replicas converge if writes stop	Cheapest; staleness window

Quorum systems give you a dial rather than a binary. With N replicas, a write acknowledged by W of them and a read answered by R of them is guaranteed to see the latest value whenever W + R > N, because the two sets must overlap in at least one replica. Below that threshold, staleness is a tunable probability.

import random

N = 3                                   # replicas per key

def quorum_run(w, r, trials=200_000, lag_ms=20, read_delay_ms=5, seed=0):
    """One write acknowledged by W replicas, then a read answered by R.
    A replica that has not received the write yet returns the old value;
    we count how often a read misses the newest write entirely."""
    rng = random.Random(seed)
    stale = 0
    for _ in range(trials):
        acked = rng.sample(range(N), w)                  # the W that acked
        fresh = set(acked)
        for nd in range(N):                              # async replication
            # Chance a lagging replica caught up before the read arrived.
            if nd not in fresh and rng.random() < read_delay_ms / lag_ms:
                fresh.add(nd)
        contacted = rng.sample(range(N), r)
        if not (fresh & set(contacted)):                 # no fresh copy seen
            stale += 1
    return stale / trials

print(f"QUORUM BEHAVIOUR WITH N = {N} REPLICAS")
print(f"  {'W':>2} {'R':>2}  {'W+R>N':>6}  {'stale reads':>11}   profile")
for w, r, profile in [(1, 1, "fastest, eventually consistent"),
                      (1, 2, "cheap writes, one hop short"),
                      (1, 3, "slow reads, read-your-writes"),
                      (2, 2, "balanced quorum (the usual default)"),
                      (3, 1, "slow writes, fast strong reads")]:
    p = quorum_run(w, r)
    print(f"  {w:>2} {r:>2}  {str(w + r > N):>6}  {p:>10.2%}   {profile}")
print("  W + R > N guarantees the read and write sets overlap, so a strong")
print("  read cannot miss the newest value. Below that line, staleness is a")
print("  probability you tune, not a bug you fix.")

  QUORUM BEHAVIOUR WITH N = 3 REPLICAS
     W  R   W+R>N  stale reads   profile
     1  1   False      49.86%   fastest, eventually consistent
     1  2   False      18.74%   cheap writes, one hop short
     1  3    True       0.00%   slow reads, read-your-writes
     2  2    True       0.00%   balanced quorum (the usual default)
     3  1    True       0.00%   slow writes, fast strong reads
    W + R > N guarantees the read and write sets overlap, so a strong
    read cannot miss the newest value. Below that line, staleness is a
    probability you tune, not a bug you fix.

For a data scientist the practical consequences are concrete. If your feature store runs on an eventually consistent store, a feature written milliseconds before inference may not be visible to the model that needs it, and the training data -- built later from the same store, when everything had converged -- will not contain that inconsistency. That is a train-serve skew caused entirely by the storage layer, and no amount of model validation will reveal it. Similarly, a metrics dashboard reading from an AP store during a partition can show two different numbers to two analysts, both "correct".

Ask, for each dataset: what is the staleness window, and does my use tolerate it? Analytics on yesterday's data tolerates seconds of staleness trivially. A fraud check that must not authorise the same card twice does not.

Key Takeaways

• Partition tolerance is not optional, so CAP reduces to: during a partition, refuse requests (CP) or serve stale data (AP).
• PACELC adds the case that matters more often -- with no partition, you still trade latency against consistency on every operation.
• Consistency is a spectrum from linearizable through causal and read-your-writes down to eventual, with cost falling along it.
• With N replicas, W + R > N forces the read and write sets to overlap, which is what makes a quorum read strong.
• Below that threshold staleness is a probability, and it can be very large: W = R = 1 missed the newest write half the time here.
• AP systems require an explicit conflict-resolution rule -- last-write -wins, vector clocks, or a CRDT -- because divergence is guaranteed.
• Eventually consistent feature stores produce train-serve skew that model-level validation cannot detect; choose the model per dataset.

🧪 Practice

1. For N = 5, list every (W, R) pair satisfying W + R > N and rank them by write latency and read latency.
2. Extend the simulation to N = 5 and plot stale-read rate against R for W = 1 and W = 2.
3. Classify these as CP or AP and justify each: a bank ledger, a shopping cart, a like counter, a DNS record, a session store.
4. Implement a last-write-wins merge and a counter CRDT, then show a write sequence where they disagree about the final value.
5. Interview: A recommendation service reads user features from an eventually consistent store. Users report that actions they just took are not reflected for a few seconds. Give two fixes at different points in the stack. (Hint: one changes which replica that user's reads go to, the other changes what the serving path reads at all.)

<a id="142-distributed-processing-frameworks"></a>

14.2 Distributed Processing Frameworks

The concepts of the previous section are delivered to you through a small set of tools, and choosing badly between them costs more than any tuning will recover; this section covers how Spark is put together, the three APIs it exposes and why only two of them are worth using, what distributed machine learning actually parallelises, the Python-native alternative that Dask offers, and the query engines that increasingly make general-purpose frameworks unnecessary.

Apache Spark Architecture

Spark exists because MapReduce wrote to disk between every stage. Its central bet was that intermediate results could be kept in memory across stages without giving up fault tolerance, which it achieves by remembering how each dataset was computed rather than checkpointing the data itself. If a partition is lost, Spark recomputes it from its recipe.

Four components, and the boundary between them explains most Spark behaviour:

  YOUR PROGRAM                          THE CLUSTER
  ------------                          -----------
  Driver
   - holds the SparkSession             Cluster manager
   - builds the logical plan              (YARN / Kubernetes / standalone)
   - Catalyst optimises it                allocates containers
   - splits stages into tasks                     |
   - schedules tasks, tracks them                 v
   - collects results             Executor 1   Executor 2   Executor 3
                                  [ task ]     [ task ]     [ task ]
                                  [ task ]     [ task ]     [ task ]
                                  cache slice  cache slice  cache slice

  The driver never touches data at scale -- except when you call collect(),
  which pulls every row into the driver's heap. That is the single most
  common way to kill a Spark job.

• The driver runs your code, plans the work, and schedules tasks. It is a single process and a single point of failure.
• Executors are JVM processes that run tasks and hold cached partitions. Each has a fixed number of slots (cores) and a memory budget.
• The cluster manager allocates executors. Spark does not manage machines.
• A task is the unit of work: one partition, one core, one stage.

Above this sits the execution model, which has three levels:

Level	Created by	How many
Job	One action (collect, write, count)	One per action
Stage	A shuffle boundary inside a job	One more than the shuffles
Task	One partition of a stage	= partition count

The distinction that matters most is narrow vs wide transformations. A narrow transformation (filter, select, withColumn) computes each output partition from exactly one input partition, so it needs no data movement and gets fused into its neighbours. A wide transformation (groupBy, join, distinct, orderBy) needs rows from many input partitions, which forces a shuffle and therefore a new stage. Counting shuffles in a plan is the fastest way to predict how expensive a job will be.

Spark is also lazy: transformations only record intent. Nothing executes until an action demands a result, which is what allows the optimiser to see the whole pipeline at once and rearrange it -- pushing filters down to the scan, collapsing projections, choosing a join strategy.

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
         .master("local[4]")           # 4 worker threads; a cluster URL in prod
         .appName("architecture-demo")
         .config("spark.sql.shuffle.partitions", 8)
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

df = (spark.range(0, 2_000_000)                       # 4 input partitions
      .withColumn("country", F.element_at(
          F.array(F.lit("IT"), F.lit("DE"), F.lit("FR")),
          (F.col("id") % 3 + 1).cast("int")))
      .withColumn("amount", (F.col("id") % 97).cast("double")))

# --- NOTHING HAS RUN YET. THESE ARE ALL TRANSFORMATIONS. ---
big = df.filter(F.col("amount") > 10)                 # narrow: no data moves
agg = big.groupBy("country").agg(F.sum("amount").alias("total"),
                                 F.count("*").alias("n"))   # wide: shuffle
print("plan built, jobs run so far:", len(spark.sparkContext.statusTracker()
                                          .getJobIdsForGroup()))

print("\nPHYSICAL PLAN (read bottom-up)")
agg.explain()                                         # still no job submitted

print("\nNOW AN ACTION")
rows = agg.orderBy("country").collect()               # this submits the job
for r in rows:
    print(f"  {r['country']}  total={r['total']:>12,.0f}  n={r['n']:>9,}")
print("  partitions in:", big.rdd.getNumPartitions(),
      " partitions after shuffle:", agg.rdd.getNumPartitions())
spark.stop()

  plan built, jobs run so far: 0

  PHYSICAL PLAN (read bottom-up)
  == Physical Plan ==
  AdaptiveSparkPlan isFinalPlan=false
  +- HashAggregate(keys=[country#1], functions=[sum(amount#2), count(1)])
     +- Exchange hashpartitioning(country#1, 8), ENSURE_REQUIREMENTS
        +- HashAggregate(keys=[country#1],
                         functions=[partial_sum(amount#2), partial_count(1)])
           +- Project [element_at([IT,DE,FR], ...) AS country#1,
                       cast((id#0L % 97) as double) AS amount#2]
              +- Filter (cast((id#0L % 97) as double) > 10.0)
                 +- Range (0, 2000000, step=1, splits=4)

  NOW AN ACTION
    DE  total=  31,621,616  n=  591,064
    FR  total=  31,621,548  n=  591,063
    IT  total=  31,621,630  n=  591,064
    partitions in: 4  partitions after shuffle: 1

Read that plan bottom-up and the whole chapter so far is visible in it. Range is the scan. Filter was pushed as low as it can go, so rows are discarded before anything else touches them. The first HashAggregate computes partial_sum and partial_count -- that is a combiner, chosen automatically. Exchange is the shuffle, and it is the only stage boundary, so this job has two stages. The final HashAggregate merges the partials.

Two details in the output are worth naming. jobs run so far: 0 after building the whole pipeline is laziness, demonstrated. And "partitions after shuffle: 1" despite spark.sql.shuffle.partitions = 8 is Adaptive Query Execution: at runtime Spark saw that the post-shuffle data was tiny and coalesced eight partitions into one, avoiding eight near-empty tasks. AQE also converts sort-merge joins to broadcast joins and splits skewed partitions when it observes real sizes -- it is on by default, and it is the reason modern Spark needs far less manual tuning than its reputation suggests.

Key Takeaways

• Spark keeps intermediate data in memory and recovers from failure by recomputing partitions from their lineage, not by checkpointing data.
• The driver plans and schedules; executors hold partitions and run tasks; the cluster manager only hands out containers.
• One action creates one job, each shuffle creates a stage boundary, and each partition of a stage is one task.
• Narrow transformations need no data movement; wide ones force a shuffle, so counting shuffles predicts cost.
• Laziness is what lets Catalyst optimise across the whole pipeline; a plan is built, not executed, until an action arrives.
• collect() moves all rows into the driver process and is a common cause of out-of-memory failures.
• Adaptive Query Execution re-plans at runtime -- coalescing partitions, switching join strategies, splitting skewed partitions.
• Read explain() bottom-up and look for Exchange nodes first.

🧪 Practice

1. Build a two-shuffle pipeline and confirm from explain() that it contains two Exchange nodes and therefore three stages.
2. Classify each of select, filter, repartition, join, withColumn, distinct, orderBy as narrow or wide.
3. Run the same aggregation with spark.sql.adaptive.enabled set to true and false, and compare the post-shuffle partition counts.
4. Add a filter before and after a groupBy in two versions of a query, then check whether the physical plans differ and explain why.
5. Interview: Your job fails with an out-of-memory error in the driver, not the executors. Name the two most likely causes. (Hint: think about which operations transfer data into the driver process, and what a broadcast variable has to pass through before it reaches executors.)

RDDs, DataFrames, and Spark SQL

Spark offers three APIs over the same engine, and they are not interchangeable -- one of them is dramatically slower and it is the one that looks most like ordinary Python.

RDDs (Resilient Distributed Datasets) are the original abstraction: a partitioned collection of opaque objects with functional operators (map, filter, reduceByKey). Spark cannot see inside your lambdas, so it cannot optimise anything. Every row becomes a Python object, and in PySpark every row crosses the JVM-to-Python boundary.

DataFrames are a partitioned collection of rows with a known schema, manipulated by declarative expressions. Because Spark understands the expressions, two layers of machinery apply:

• Catalyst, the query optimiser: predicate pushdown, projection pruning, constant folding, join reordering, join strategy selection.
• Tungsten, the execution layer: off-heap binary storage that skips JVM object overhead, cache-aware algorithms, and whole-stage code generation that compiles a chain of operators into a single tight JVM loop.

Spark SQL is the same DataFrame engine with SQL text as the front end. Identical plans, identical performance -- choose it wherever SQL expresses the transformation more clearly.

The decisive consequence for a Python user: a DataFrame expression never leaves the JVM, while an RDD lambda or a Python UDF forces every row through a serialisation round trip into a Python worker process.

  WHERE THE WORK HAPPENS

  DataFrame expression                Python UDF / RDD lambda
  --------------------                -----------------------
   JVM: read binary row                JVM: read binary row
   JVM: generated loop                 serialise row  ---> pickle
   JVM: write binary row               Python worker: your function
                                       serialise result <--- pickle
                                       JVM: write binary row

  Two extra serialisations and a process hop, per row.
  Pandas UDFs (vectorised, Arrow-based) pay it once per BATCH instead,
  which recovers most of the gap when a UDF is unavoidable.

import time
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import DoubleType

spark = (SparkSession.builder.master("local[4]").appName("api-layers")
         .config("spark.sql.shuffle.partitions", 8).getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

N = 8_000_000
df = (spark.range(N)
      .withColumn("country", F.element_at(
          F.array(F.lit("IT"), F.lit("DE"), F.lit("FR")),
          (F.col("id") % 3 + 1).cast("int")))
      .withColumn("amount", (F.col("id") % 97).cast("double"))).cache()
df.count()                            # materialise the cache before timing

def timed(label, fn):
    t = time.perf_counter(); out = fn(); dt = time.perf_counter() - t
    print(f"  {label:34s} {dt:6.2f}s")
    return out, dt

# 1. RDD API: your Python lambdas, one object per row, no optimiser.
def rdd_way():
    return (df.rdd
            .map(lambda r: (r["country"], (r["amount"], 1)))
            .reduceByKey(lambda a, b: (a[0] + b[0], a[1] + b[1]))
            .mapValues(lambda t: t[0] / t[1])
            .collect())

# 2. DataFrame API: declarative, compiled by Catalyst to JVM bytecode.
def df_way():
    return df.groupBy("country").agg(F.avg("amount")).collect()

# 3. Spark SQL: same plan, different surface syntax.
df.createOrReplaceTempView("events")
def sql_way():
    return spark.sql("SELECT country, avg(amount) FROM events "
                     "GROUP BY country").collect()

print(f"MEAN AMOUNT PER COUNTRY OVER {N:,} ROWS")
_, t_rdd = timed("RDD map/reduceByKey", rdd_way)
_, t_df  = timed("DataFrame groupBy/agg", df_way)
_, t_sql = timed("Spark SQL", sql_way)
print(f"  DataFrame is {t_rdd / t_df:.1f}x faster than the RDD version")

# --- PYTHON UDFS BREAK THE OPTIMISER'S GRIP. ---
@F.udf(DoubleType())
def py_bucket(x):                     # runs in a Python worker, row by row
    return float(x // 10 * 10)

def udf_way():
    return df.withColumn("b", py_bucket("amount")).groupBy("b").count().collect()

def native_way():
    return (df.withColumn("b", F.floor(F.col("amount") / 10) * 10)
            .groupBy("b").count().collect())

print("\nBUCKETING THE SAME COLUMN TWO WAYS")
_, t_udf = timed("Python UDF", udf_way)
_, t_nat = timed("native Spark expression", native_way)
print(f"  the UDF costs {t_udf / t_nat:.1f}x more: every row is serialised to")
print("  a Python process and back, and Catalyst cannot see inside it")
spark.stop()

  MEAN AMOUNT PER COUNTRY OVER 8,000,000 ROWS
    RDD map/reduceByKey                 19.00s
    DataFrame groupBy/agg                1.49s
    Spark SQL                            0.94s
    DataFrame is 12.8x faster than the RDD version

  BUCKETING THE SAME COLUMN TWO WAYS
    Python UDF                           4.89s
    native Spark expression              0.76s
    the UDF costs 6.4x more: every row is serialised to
    a Python process and back, and Catalyst cannot see inside it

Absolute times depend on the machine; the ratios are the lesson. Identical logic and identical data ran an order of magnitude apart purely because of which API expressed it.

	RDD	DataFrame	Spark SQL
Schema known to engine	No	Yes	Yes
Catalyst optimisation	None	Full	Full
Tungsten binary format	No	Yes	Yes
PySpark serialisation cost	Per row	None	None
Compile-time type safety	Yes (Scala)	No	No
Right choice when	Truly unstructured data, custom partitioners	Almost always	Logic reads better as SQL

When a UDF is genuinely necessary -- a library call, a model prediction, a parser -- reach for a pandas UDF rather than a scalar Python UDF. It receives an Arrow batch as a pandas.Series, so the serialisation cost is paid once per batch of thousands of rows instead of once per row, and vectorised NumPy code runs over the whole batch.

Key Takeaways

• RDDs hold opaque objects and defeat the optimiser; DataFrames carry a schema the engine can reason about.
• Catalyst rewrites plans (pushdown, pruning, join selection) and Tungsten executes them on off-heap binary data with generated code.
• Spark SQL and the DataFrame API compile to the same plans, so the choice is purely about readability.
• In PySpark, every RDD lambda and scalar UDF serialises each row into a Python process and back.
• Measured here: DataFrame 12.8x faster than the equivalent RDD job, and a native expression 6.4x faster than the same logic in a UDF.
• Use pandas (Arrow) UDFs when a UDF is unavoidable -- the cost becomes per batch rather than per row.
• Reach for RDDs only for genuinely unstructured data or custom partitioning logic.

🧪 Practice

1. Rewrite a scalar Python UDF as a native Spark expression and measure both on at least 5 million rows.
2. Convert the same UDF into a pandas UDF and compare all three timings.
3. Write one aggregation in the DataFrame API and the same one in Spark SQL, then diff their physical plans.
4. Take an RDD job using groupByKey and rewrite it with reduceByKey, explaining which one shuffles less and why.
5. Interview: A colleague says "we use RDDs because we need custom logic per row". What do you ask, and what alternatives do you propose? (Hint: separate the logic that must be Python from the logic that is really an expression, and consider where the batch boundary could go.)

Distributed Machine Learning Libraries

"Distributed machine learning" conflates two entirely different problems, and naming which one you have determines every tool choice that follows.

Data parallelism: the model fits on one machine, the data does not. Each worker holds a slice of the data, computes gradients or statistics on it, and the partial results are aggregated. This is the common case in tabular data science, and it is what Spark MLlib does.

Model parallelism: the model itself does not fit on one machine, so its layers or parameters are split across devices. This is a deep learning concern -- large transformers -- and it lives in the frameworks covered in Chapter 12, not here.

  DATA PARALLEL (one model, many data slices)

    slice 1 --> [full model copy] --> partial gradient --\
    slice 2 --> [full model copy] --> partial gradient ---+-> average
    slice 3 --> [full model copy] --> partial gradient --/     |
                                                              v
                       every copy updated with the same new parameters

  MODEL PARALLEL (one data batch, model split across devices)

    batch --> [layers 1-8 on GPU0] --> [layers 9-16 on GPU1] --> output
              activations cross the device boundary every step

Spark MLlib provides the data-parallel case with an API deliberately shaped like scikit-learn's: Transformer (has transform), Estimator (has fit, returns a Transformer), and Pipeline (a sequence of stages, itself an Estimator). The reason to care about the Pipeline abstraction is not tidiness -- it is that the fitted pipeline contains the fitted preprocessing, so the exact category-to-index mapping learned at training time is applied at serving time. Hand-rolled preprocessing is the single largest source of train-serve skew, and pipelines are the structural fix.

from pyspark.sql import SparkSession, functions as F
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator

spark = (SparkSession.builder.master("local[4]").appName("mllib")
         .config("spark.sql.shuffle.partitions", 8).getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

N = 500_000
df = (spark.range(N)
      .withColumn("country", F.element_at(
          F.array(F.lit("IT"), F.lit("DE"), F.lit("FR")),
          (F.col("id") % 3 + 1).cast("int")))
      .withColumn("tenure", (F.col("id") % 60).cast("double"))
      .withColumn("spend", (F.rand(7) * 500))
      # Churn is driven by short tenure, with noise on top.
      .withColumn("label", (F.when(F.col("tenure") < 6, F.rand(1) * 0.6 + 0.3)
                            .otherwise(F.rand(2) * 0.4) > 0.5).cast("int")))

train, test = df.randomSplit([0.8, 0.2], seed=42)

pipe = Pipeline(stages=[
    StringIndexer(inputCol="country", outputCol="country_ix"),
    OneHotEncoder(inputCols=["country_ix"], outputCols=["country_oh"]),
    # MLlib estimators take ONE vector column, never a list of columns.
    VectorAssembler(inputCols=["tenure", "spend", "country_oh"],
                    outputCol="features"),
    LogisticRegression(maxIter=20, regParam=0.01),
])

model = pipe.fit(train)          # each stage fitted in order, distributed
pred = model.transform(test)     # the SAME fitted mappings are reapplied
auc = BinaryClassificationEvaluator(metricName="areaUnderROC").evaluate(pred)
lr = model.stages[-1]
print(f"rows train / test: {train.count():,} / {test.count():,}")
print(f"AUC {auc:.4f}")
print("coefficients:", "  ".join(f"{c:+.4f}" for c in lr.coefficients))
print("iterations run:", lr.summary.totalIterations)
print("objective history (first 5):",
      [round(v, 4) for v in lr.summary.objectiveHistory[:5]])
spark.stop()

  rows train / test: 400,052 / 99,948
  AUC 0.9827
  coefficients: -0.1290  +0.0000  +0.2141  +0.0847
  iterations run: 11
  objective history (first 5): [0.244, 0.1793, 0.1654, 0.1551, 0.1524]

The objectiveHistory is worth noticing: every one of those 11 iterations was a full pass over 400,000 rows, with a gradient aggregation -- a shuffle-like reduction -- at each step. That is why iterative algorithms were the workload that justified Spark's in-memory design, and why caching the training set before fit is not optional.

The distributed ML landscape, and when each is the right answer:

Tool	Parallelism	Use when
Spark MLlib	Data	Data already in Spark; standard models suffice
XGBoost / LightGBM (distributed)	Data	Gradient boosting on data exceeding one machine
Dask-ML	Data	Python-native stack; scikit-learn compatibility matters
Ray Train / Tune	Data + tasks	Heterogeneous workloads, RL, large-scale HPO
PyTorch DDP / FSDP	Data / model	Deep learning on multiple GPUs
scikit-learn on one machine	None	The data fits -- which is most of the time

That last row is the one most often skipped. MLlib's model catalogue is narrower than scikit-learn's, its hyperparameter tuning is coarser, and it carries per-iteration coordination overhead. A very common and very effective pattern is: aggregate and sample at scale in Spark, then train on one machine. Ten million rows of raw events often reduce to 200,000 feature rows, and 200,000 rows belong in scikit-learn, LightGBM, or CatBoost -- where you get better models, faster iteration, and real debugging.

There is also a third kind of parallelism that is easy to forget and cheap to exploit: embarrassingly parallel model fitting. Training one model per store, per region, or per SKU is thousands of independent small fits, not one distributed fit. Distribute the tasks, not the algorithm -- a groupBy(...).applyInPandas(...) in Spark, or dask.delayed, does this with no distributed-training machinery at all.

Key Takeaways

• Distinguish data parallelism (model fits, data does not) from model parallelism (the model itself does not fit); tabular work is almost always the former.
• Spark MLlib is data-parallel, with a scikit-learn-shaped Transformer/Estimator/Pipeline API.
• Pipelines matter because the fitted preprocessing travels with the fitted model, which structurally prevents a major class of train-serve skew.
• MLlib estimators consume a single assembled vector column, not a list of feature columns.
• Iterative training does a full distributed pass per iteration, so cache the training set before fitting.
• Aggregating in Spark and then training on one machine is usually the better pipeline; the feature table is far smaller than the event log.
• Many-small-models problems are task-parallel, not data-parallel; use grouped pandas application or delayed tasks instead.

🧪 Practice

1. Build an MLlib pipeline with an indexer, encoder, assembler, and classifier, and verify that a category unseen in training is handled at transform time.
2. Fit the same model on a sample small enough for scikit-learn and compare AUC and wall-clock time against the MLlib run.
3. Use groupBy(...).applyInPandas(...) to fit one model per group and return a row of coefficients per group.
4. Time an MLlib fit with and without caching the training DataFrame, and explain the gap using the iteration count.
5. Interview: You need a model per country for 180 countries, each with up to 2 million rows. Data-parallel or task-parallel, and how would you implement it? (Hint: ask whether any single country's data exceeds one machine, and what that answer implies about which axis to distribute.)

Dask and Parallel Python

Dask answers a question Spark does not: what if you want to scale out while staying inside the Python scientific stack? Spark is a JVM engine you talk to across a boundary. Dask is Python all the way down, which changes what is convenient and what is possible.

Its abstractions mirror the libraries you already use:

• dask.dataframe -- many pandas DataFrames, partitioned along the index.
• dask.array -- many NumPy arrays, in a blocked grid.
• dask.bag -- a parallel sequence for semi-structured data.
• dask.delayed -- wrap any function call as a lazy graph node. This is the escape hatch that makes arbitrary Python parallelisable.

Under all of them is one mechanism: a task graph, a directed acyclic graph of Python callables and their dependencies, executed by a scheduler that respects the edges. Because the graph is an ordinary Python data structure, you can inspect it, which is a genuine pedagogical advantage over Spark's opaque JVM plans.

  A DASK DATAFRAME IS A ROW-WISE STACK OF PANDAS DATAFRAMES

    partition 0   day=00   [ pandas DataFrame ]  --\
    partition 1   day=01   [ pandas DataFrame ]  ---\
    partition 2   day=02   [ pandas DataFrame ]  ----+-- groupby.agg
    ...                                             /   (tree reduction)
    partition 11  day=11   [ pandas DataFrame ]  --/

  Any operation along the partition axis (groupby on a non-index column,
  set_index, merge on a non-index column) requires a shuffle, exactly as
  in Spark. Everything else is per-partition pandas.

import os, tempfile, time
import numpy as np, pandas as pd
import dask.dataframe as dd
from dask.distributed import Client

# --- 12 PARQUET PART FILES: THE UNIT OF PARALLELISM IS THE PARTITION. ---
d = tempfile.mkdtemp()
rng = np.random.default_rng(0)
for i in range(12):
    n = 500_000
    pd.DataFrame({
        "day": np.full(n, i),
        "country": rng.choice(["IT", "DE", "FR", "ES"], n),
        "amount": rng.exponential(40, n).round(2),
    }).to_parquet(os.path.join(d, f"part-{i:02d}.parquet"), index=False)
print("files:", len(os.listdir(d)), " total rows: 6,000,000")

# processes=True in production; threads keep this example self-contained.
client = Client(n_workers=4, threads_per_worker=1, processes=False,
                dashboard_address=None)

ddf = dd.read_parquet(d)                       # lazy: reads only the schema
print("partitions:", ddf.npartitions, " columns:", list(ddf.columns))

res = (ddf[ddf.amount > 5]
       .groupby("country")
       .amount.agg(["mean", "count"]))         # still lazy, a task graph
t = time.perf_counter()
out = res.compute()                            # now it runs
print(f"\ncomputed in {time.perf_counter() - t:.2f}s")
print(out.sort_index().round(3).to_string())

# Column projection is pushed into the Parquet reader.
small = dd.read_parquet(d, columns=["country", "amount"])
print("\ncolumns actually read:", list(small.columns))

# --- THE GRAPH IS DATA YOU CAN INSPECT. ---
print("\ntasks in the graph for the aggregation:",
      len(dict(res.optimize().dask)))

# --- map_partitions RUNS ORDINARY PANDAS/NUMPY CODE PER PARTITION. ---
def p95_per_partition(pdf):
    """Receives a real pandas DataFrame -- any library that works on one
    partition works here, which is Dask's main advantage over Spark for
    Python-native code."""
    return pd.DataFrame({"day": [pdf["day"].iloc[0]],
                         "p95": [pdf["amount"].quantile(0.95)]})

meta = pd.DataFrame({"day": pd.Series(dtype="int64"),
                     "p95": pd.Series(dtype="float64")})
per_day = ddf.map_partitions(p95_per_partition, meta=meta).compute()
print(per_day.head(4).round(2).to_string(index=False))
client.close()

  files: 12  total rows: 6,000,000
  partitions: 12  columns: ['day', 'country', 'amount']

  computed in 0.62s
             mean    count
  country
  DE       44.983  1323973
  ES       44.970  1323355
  FR       45.027  1323516
  IT       45.006  1324046

  columns actually read: ['country', 'amount']

  tasks in the graph for the aggregation: 15
   day    p95
     0 119.83
     1 119.73
     2 119.86
     3 120.19

Two Dask-specific traps are worth stating plainly.

meta is mandatory and easy to get wrong. Dask must know the output schema before running anything, so map_partitions takes an empty frame describing the result. Get its dtypes wrong and you get a confusing failure deep in execution, or silently wrong dtypes.

Partition size is your responsibility. Dask does not choose it for you. Aim for roughly 100 MB of in-memory data per partition: too large and workers die of memory exhaustion, too small and scheduler overhead dominates. A common symptom of a badly partitioned Dask job is workers being killed and restarted in a loop.

	Dask	Spark
Language	Python	JVM (Scala/Java), Python via a bridge
API	Mirrors pandas/NumPy	Its own DataFrame API + SQL
Arbitrary Python cost	Native, free	Serialisation per row or per batch
Query optimiser	Limited (column/predicate pushdown)	Catalyst, mature
SQL support	Via add-ons	First-class
Scale sweet spot	One machine to ~100 nodes	Tens to thousands of nodes
Ecosystem maturity at scale	Younger	Extremely battle-tested
Best for	NumPy/pandas/scikit-learn workloads, custom algorithms	Large ETL, SQL, mixed-language teams

The honest summary: if your workload is SQL-shaped ETL over very large tables, Spark. If it is Python-shaped -- array mathematics, scikit-learn, custom per-group logic -- Dask, and you keep your debugger. And if the data fits in memory on one machine, neither: pandas or Polars will beat both.

Key Takeaways

• Dask scales the Python scientific stack by partitioning into many pandas DataFrames or NumPy blocks.
• Everything compiles to a task graph of Python callables, which is an inspectable data structure.
• dask.delayed parallelises arbitrary Python functions, which is the escape hatch Spark lacks.
• Operations that cross the partition axis (non-index groupby, set_index, non-index merge) shuffle, just as in Spark.
• meta must describe the output schema of map_partitions correctly, or failures appear far from the cause.
• Choose partition sizes yourself, around 100 MB in memory; repeated worker restarts usually mean partitions are too large.
• Spark for SQL-shaped ETL at large scale, Dask for Python-shaped computation, and neither when one machine will do.

🧪 Practice

1. Write a dataset as 4, 12, and 200 Parquet files, then compare Dask aggregation time across the three partition counts.
2. Use map_partitions to compute a per-partition statistic, deliberately supply a wrong meta dtype, and observe how the error surfaces.
3. Parallelise a loop of independent model fits with dask.delayed and confirm the graph has one node per fit.
4. Compare ddf.groupby("col").mean() on an index column versus a non-index column and explain the runtime difference.
5. Interview: A Dask job repeatedly kills and restarts its workers with memory warnings, though the dataset is only 30 GB across 40 workers. What do you check first? (Hint: total size is not the quantity that has to fit anywhere -- ask what a single task holds at once.)

Query Engines for Large Data

For a large share of analytics, a general-purpose distributed framework is the wrong tool, because the workload is a SQL query and there are engines built for nothing else. They have absorbed decades of database research, and they routinely beat hand-written Spark jobs on the queries they cover.

Three architectural families:

Embedded vectorised engines (DuckDB, Polars) run in your process, use all cores, and process data in cache-sized column batches. No cluster, no scheduler, no network. On a modern machine they handle hundreds of gigabytes of Parquet, and their startup cost is milliseconds. This category has reclaimed most of the range where Spark used to be the reflex.

Distributed MPP query engines (Trino, Presto, Spark SQL, Dremio) split a SQL query across nodes that read from object storage. They exist for data genuinely beyond one machine and for federating across sources, and their central value is that they decouple storage from compute.

Cloud data warehouses (BigQuery, Snowflake, Redshift) are managed MPP engines with storage, metadata, and scaling handled for you. You trade control and per-query cost for the absence of operational work.

Whichever family, the same three optimisations determine whether a query takes seconds or hours:

1. Projection pruning -- read only the columns the query mentions. Free with columnar formats; impossible with CSV or JSON.
2. Partition pruning -- skip whole directories using predicates on partition columns. This is why WHERE day = '2026-09-28' on a day-partitioned table is fast and WHERE amount > 5 is not.
3. Predicate pushdown with statistics -- inside a Parquet file, every row group stores per-column min/max, so a row group whose range cannot satisfy the predicate is skipped without decompression.

import os, time, tempfile
import numpy as np, pandas as pd, duckdb

# --- HIVE-STYLE PARTITIONED LAYOUT: THE DIRECTORY NAME IS AN INDEX. ---
root = tempfile.mkdtemp()
rng = np.random.default_rng(0)
for day in range(1, 13):
    p = os.path.join(root, f"day={day:02d}")      # the partition column
    os.makedirs(p)
    n = 400_000
    pd.DataFrame({
        "country": rng.choice(["IT", "DE", "FR", "ES"], n),
        "amount": rng.exponential(40, n).round(2),
        "user_id": rng.integers(0, 200_000, n),
    }).to_parquet(os.path.join(p, "data.parquet"), index=False)

total_mb = sum(os.path.getsize(os.path.join(dp, f))
               for dp, _, fs in os.walk(root) for f in fs) / 1024 ** 2
print(f"dataset: 12 day-partitions, 4,800,000 rows, {total_mb:.1f} MB on disk")

con = duckdb.connect()                            # in-process, no server
glob = f"'{root}/**/*.parquet'"

def timed(label, sql):
    t = time.perf_counter()
    out = con.sql(sql).fetchall()
    print(f"  {label:38s} {time.perf_counter() - t:6.3f}s  -> {out[0]}")

print("\nSAME QUESTION, THREE AMOUNTS OF WORK")
timed("scan every column, every partition",
      f"SELECT count(*), sum(amount), sum(user_id) "
      f"FROM read_parquet({glob}, hive_partitioning=true)")
timed("project: only the column needed",
      f"SELECT count(*), sum(amount) "
      f"FROM read_parquet({glob}, hive_partitioning=true)")
timed("project + partition prune (day=03)",
      f"SELECT count(*), sum(amount) "
      f"FROM read_parquet({glob}, hive_partitioning=true) WHERE day='03'")

print("\nWHAT THE PLANNER DID WITH THE PARTITION FILTER")
plan = con.sql(f"EXPLAIN SELECT sum(amount) FROM read_parquet({glob},"
               f" hive_partitioning=true) WHERE day='03'").fetchall()[0][1]
for line in plan.splitlines():
    if any(k in line for k in ("File Filters", "Scan", "Projection",
                               "Filters", "Total Files Read")):
        print("   ", line.strip())
con.close()

  dataset: 12 day-partitions, 4,800,000 rows, 33.0 MB on disk

  SAME QUESTION, THREE AMOUNTS OF WORK
    scan every column, every partition      0.074s
      -> (4800000, 192048740.59999803, 480062630010)
    project: only the column needed         0.013s
      -> (4800000, 192048740.59999803)
    project + partition prune (day=03)      0.006s
      -> (400000, 15986678.359999856)

  WHAT THE PLANNER DID WITH THE PARTITION FILTER
      │    Projections: amount    │
      │       File Filters:       │
      │    Scanning Files: 1/12   │

Scanning Files: 1/12 is the whole topic in one line. The engine never opened the other eleven files, because the directory name told it they could not contain matching rows. Dropping one unused column gave 5.7x, and pruning eleven of twelve partitions gave a further 2.2x on top -- 12x in total, from the physical layout of the data rather than from anything in the query. It scales with the partition count: on a three-year day-partitioned table, a one-day query touches roughly 1/1,000 of the files.

This also explains the most consequential modelling decision in an analytics store: partition by the column your queries filter on, almost always a date, and avoid partitioning by anything high-cardinality (a user_id partition column produces millions of tiny files and the small-files problem of the previous section).

Engine family	Deploy	Sweet spot	Watch out for
DuckDB / Polars	In your process	Up to hundreds of GB Parquet	Single machine's RAM and cores
Trino / Presto	Cluster you run	Multi-TB, federated sources	Operations, no built-in storage
Spark SQL	Cluster you run	ETL plus ML in one system	Startup cost, tuning surface
BigQuery / Snowflake	Managed	Elastic, zero ops	Per-query or per-credit cost

The decision procedure is short: does one machine's RAM and core count cover the query after projection and partition pruning? If yes, use an embedded engine and skip the cluster entirely. If no, or if the data must stay in object storage while many teams query it, use a distributed engine.

Key Takeaways

• Most analytics workloads are SQL, and dedicated query engines beat general frameworks on them.
• Embedded vectorised engines (DuckDB, Polars) handle hundreds of GB of Parquet in-process, with millisecond startup.
• Distributed MPP engines exist for data beyond one machine and for federating across sources, decoupling storage from compute.
• Projection pruning, partition pruning, and row-group statistics decide query cost far more than engine choice does.
• Partition on the column queries filter on -- usually a date -- and never on a high-cardinality column.
• Scanning Files: 1/12 in a plan is the evidence that pruning worked; learn to look for the equivalent line in your engine.
• Check whether one machine suffices after pruning before reaching for a cluster.

🧪 Practice

1. Write the same dataset partitioned by day and unpartitioned, then time a single-day query against both.
2. Run EXPLAIN on a filtered query and identify which files or row groups the engine decided to skip.
3. Partition a dataset by a high-cardinality column, count the files produced, and measure the effect on a full-table scan.
4. Compare DuckDB against Spark SQL on the same 5 GB Parquet aggregation, including process startup time in both measurements.
5. Interview: A dashboard query over a day-partitioned table filters on event_timestamp rather than the day partition column and is slow. What is happening and how do you fix it? (Hint: the engine prunes on the partition column only, so ask whether the filter can be expressed in terms the directory layout exposes.)

<a id="143-streaming-data"></a>

14.3 Streaming Data

Batch processing asks a question of a finite dataset; streaming asks a standing question of data that never ends, and that difference reshapes correctness rather than just performance; this section covers what changes when the input is unbounded, why the timestamp on a record matters more than the clock on the server, how to cut an endless stream into windows you can aggregate, the log systems that carry the events, and how to compute features online without teaching your model about the future.

Stream Processing Concepts

In batch processing the dataset is finite and available, so a computation runs once and produces one answer. In stream processing the input never ends, which means there is no moment at which the answer is final. A streaming computation produces a sequence of answers, each correct as of the records it has seen.

That single change has consequences that are not about speed at all:

	Batch	Stream
Input	Bounded, complete	Unbounded, arriving
Completeness	Known	Never; you decide when to stop waiting
State	Held for the job's duration	Held indefinitely, must be bounded
Failure recovery	Re-run the job	Resume from a checkpoint and offset
"Correct" result	One	One per moment, all correct
Reprocessing	Trivial	Requires replayable input

The vocabulary is small and worth fixing precisely:

• Source: where records come from (a Kafka topic, a change feed).
• Operator: a transformation applied per record or per window.
• State: what the operator remembers between records -- a running sum, a set of seen ids, a join buffer.
• Sink: where results go (a database, a topic, a dashboard).
• Checkpoint: a consistent snapshot of all operator state plus the input offsets it corresponds to. This pairing is what makes recovery correct.
• Backpressure: the mechanism by which a slow sink slows the source rather than causing unbounded memory growth.

Two engine designs exist. Micro-batching (Spark Structured Streaming) collects records for a short interval and runs a small batch job, giving latency in the hundreds of milliseconds to seconds and reusing the entire batch engine. Record-at-a-time (Flink, Kafka Streams) processes each record as it arrives, reaching single-digit-millisecond latency at the cost of a separate execution model. For most analytics and machine learning, micro-batching is entirely sufficient, and its operational simplicity is worth more than the latency difference.

The concept that causes the most confusion is delivery semantics, so it is worth deriving rather than memorising. A pipeline must both write a result and record how far it has read. Whichever it does first determines what a crash between them costs.

# --- A BOUNDED TABLE VS AN UNBOUNDED STREAM, SAME QUESTION. ---
# Batch: the data is all here, so "the answer" is well defined.
events = [("u1", 10.0), ("u2", 4.0), ("u1", 6.0), ("u3", 9.0)]
print("BATCH: one answer, computed once")
totals = {}
for u, a in events:
    totals[u] = totals.get(u, 0.0) + a
print("  ", dict(sorted(totals.items())))

# Stream: the data never ends, so the answer is a SEQUENCE of answers, each
# correct as of the records seen so far.
print("\nSTREAM: an answer per record, each correct at its moment")
state = {}
for i, (u, a) in enumerate(events, 1):
    state[u] = state.get(u, 0.0) + a
    print(f"   after record {i}: {dict(sorted(state.items()))}")

# --- DELIVERY SEMANTICS, MADE CONCRETE. ---
def pipeline(records, semantics, crash_before_commit_of=3):
    """Process records and crash exactly once, after the work but before the
    offset is committed. What the sink ends up holding depends entirely on
    the ORDER of `commit offset` and `write to sink`."""
    sink, offset, crashed = [], 0, False
    while offset < len(records):
        rec = records[offset]
        if semantics == "at-most-once":
            offset += 1                       # commit FIRST, then do the work
            if offset - 1 == crash_before_commit_of and not crashed:
                crashed = True                # died before writing: lost
                continue
            sink.append(rec)
            continue
        if semantics == "exactly-once":
            # Keyed upsert: a replay overwrites rather than appends.
            sink = [r for r in sink if r[0] != rec[0]] + [rec]
        else:                                 # at-least-once: blind append
            sink.append(rec)
        if offset == crash_before_commit_of and not crashed:
            crashed = True                    # work done, offset NOT committed
            continue                          # restart replays this record
        offset += 1                           # commit AFTER the work
    return sink

recs = [(f"k{i}", i * 10) for i in range(6)]
print("\nDELIVERY SEMANTICS AFTER ONE CRASH MID-STREAM")
for sem in ("at-most-once", "at-least-once", "exactly-once"):
    out = pipeline(recs, sem)
    keys = [k for k, _ in out]
    dupes = len(keys) - len(set(keys))
    missing = len(set(k for k, _ in recs) - set(keys))
    print(f"  {sem:15s} sink has {len(out)} rows,"
          f" {dupes} duplicate(s), {missing} lost")
print("  Exactly-once is not magic: it is at-least-once delivery plus a sink")
print("  whose writes are idempotent or transactional.")

  BATCH: one answer, computed once
     {'u1': 16.0, 'u2': 4.0, 'u3': 9.0}

  STREAM: an answer per record, each correct at its moment
     after record 1: {'u1': 10.0}
     after record 2: {'u1': 10.0, 'u2': 4.0}
     after record 3: {'u1': 16.0, 'u2': 4.0}
     after record 4: {'u1': 16.0, 'u2': 4.0, 'u3': 9.0}

  DELIVERY SEMANTICS AFTER ONE CRASH MID-STREAM
    at-most-once    sink has 5 rows, 0 duplicate(s), 1 lost
    at-least-once   sink has 7 rows, 1 duplicate(s), 0 lost
    exactly-once    sink has 6 rows, 0 duplicate(s), 0 lost
    Exactly-once is not magic: it is at-least-once delivery plus a sink
    whose writes are idempotent or transactional.

Three semantics, one line of difference in the code. At-most-once commits the offset before doing the work, so a crash loses a record -- acceptable for approximate metrics, never for money. At-least-once does the work first, so a crash duplicates a record; it is the default almost everywhere. Exactly-once is at-least-once plus a sink whose writes can absorb a replay: a keyed upsert, a transaction covering both the write and the offset, or deduplication on an event id. Notice that the third case required no change to delivery at all, only to how the sink writes. Whenever a vendor claims exactly-once, the question to ask is which of those mechanisms is providing it and whether your sink supports it.

Key Takeaways

• A streaming computation has no final answer; it produces a sequence of answers, each correct as of what has arrived.
• State must be explicitly bounded, because the input never ends.
• A checkpoint pairs operator state with the input offsets it corresponds to; that pairing is what makes recovery correct.
• Backpressure propagates slowness upstream instead of letting memory grow without limit.
• Micro-batching reuses the batch engine for second-scale latency; record-at-a-time engines reach milliseconds with more machinery.
• At-most-once loses records, at-least-once duplicates them, and exactly-once requires an idempotent or transactional sink.
• Reprocessing history requires a replayable source, which is the main architectural reason to put a log in front of a stream processor.

🧪 Practice

1. Rewrite a batch aggregation as an incremental one that updates a running state per record, and print the state after each record.
2. Modify the crash simulation so the sink deduplicates on an event id, and show that at-least-once delivery now yields exactly-once results.
3. List three sinks you use and classify each as idempotent, transactional, or neither.
4. For a stream of 50,000 events per second with per-user state of 200 bytes and 4 million active users, compute the state size and say what bounds it.
5. Interview: A stakeholder asks for "exactly-once processing". What do you need to know before agreeing it is achievable? (Hint: the guarantee is a property of the whole path from source to sink, so ask what the sink can do when it sees the same record twice.)

Event Time vs Processing Time

Every streaming record has at least two timestamps, and confusing them produces results that look fine and are wrong.

• Event time: when the thing happened, stamped at the source. A property of the world.
• Ingestion time: when the record reached the broker.
• Processing time: when your operator looked at it. A property of your infrastructure.

The gap between event time and processing time is never zero and never constant. A phone loses signal in a tunnel and uploads forty events at once, twenty minutes late. A retry queue drains after an outage. A batch uploader runs every fifteen minutes. So records arrive out of order in event time, and the disorder is unbounded in principle.

Why does the choice matter? Because processing-time results are not reproducible. Restart the job, and the windows land differently. Replay yesterday's data to backfill, and every record lands in one window -- the one containing "now". Event-time results are reproducible by construction: the same input gives the same output regardless of when or how fast you process it.

  THE SAME FIVE EVENTS, TWO TIMELINES

  event time   e1   e2        e3   e4        e5
               |    |         |    |         |
               0   10        30   40        55         (what happened)

  arrival      e1   e2   e4   e5        e3
               |    |    |    |         |
               2   12   43   57        95              (what you observed)

  Windowing on arrival puts e3 in a later window than e4, though it
  happened earlier. The count per window becomes a fact about your
  network, not about your users.

The mechanism that makes event-time processing possible is the watermark: an assertion by the engine that no record with an event time earlier than W is expected any more. It is usually computed as max event time seen so far - allowed lateness. When the watermark passes the end of a window, that window is complete and can be emitted and its state discarded.

Allowed lateness is a direct, unavoidable trade: longer lateness means more complete results, more state held, and later emission. There is no setting that gives all three.

import random
from collections import defaultdict

rng = random.Random(3)
WINDOW = 60          # seconds
LATE_P = 0.12        # share of events arriving badly out of order

# Each event knows WHEN IT HAPPENED (event_time) and arrives later.
stream = []
for et in range(0, 300):                      # one event per second, 5 minutes
    delay = rng.expovariate(1 / 2) if rng.random() > LATE_P \
            else rng.uniform(30, 150)         # stragglers: offline phones
    stream.append({"event_time": et, "arrival_time": et + delay, "value": 1})
stream.sort(key=lambda e: e["arrival_time"])   # the order the system sees

def window_of(t):  return int(t // WINDOW) * WINDOW

by_event = defaultdict(int)
by_processing = defaultdict(int)
for e in stream:
    by_event[window_of(e["event_time"])] += e["value"]
    by_processing[window_of(e["arrival_time"])] += e["value"]

print("COUNTS PER 60s WINDOW: TWO DEFINITIONS OF TIME")
print(f"  {'window':>8} {'by event time':>14} {'by processing time':>19}")
for w in sorted(set(by_event) | set(by_processing)):
    print(f"  {w:>8} {by_event.get(w, 0):>14} {by_processing.get(w, 0):>19}")
print("  Event time reproduces what happened. Processing time reproduces")
print("  when your pipeline happened to be looking, which is not a property")
print("  of the business at all -- it changes if you restart the job.")

# --- WATERMARKS: DECIDING WHEN A WINDOW IS DONE. ---
def run_with_watermark(stream, allowed_lateness):
    """Watermark = max event time seen so far - allowed_lateness. A window
    closes when the watermark passes its end; anything later is dropped."""
    watermark, closed, counts, dropped = -1, set(), defaultdict(int), 0
    max_et = -1
    for e in stream:
        max_et = max(max_et, e["event_time"])
        watermark = max_et - allowed_lateness
        w = window_of(e["event_time"])
        if w in closed:
            dropped += 1                       # arrived after its window shut
            continue
        counts[w] += e["value"]
        for cand in list(counts):              # close everything now complete
            if cand + WINDOW <= watermark:
                closed.add(cand)
    return counts, dropped

print("\nWATERMARK TRADE-OFF: COMPLETENESS VS LATENCY")
truth = sum(by_event.values())
for lateness in (0, 10, 30, 60, 180):
    counts, dropped = run_with_watermark(stream, lateness)
    got = sum(counts.values())
    print(f"  lateness {lateness:>3}s -> {got:>3}/{truth} counted"
          f" ({got / truth:5.1%}), {dropped:>3} dropped")
print("  Zero lateness closes windows immediately and loses every straggler.")
print("  Large lateness is complete but holds state and delays results.")

  COUNTS PER 60s WINDOW: TWO DEFINITIONS OF TIME
      window  by event time  by processing time
           0             60                  56
          60             60                  55
         120             60                  54
         180             60                  67
         240             60                  58
         300              0                   9
         360              0                   1
    Event time reproduces what happened. Processing time reproduces
    when your pipeline happened to be looking, which is not a property
    of the business at all -- it changes if you restart the job.

  WATERMARK TRADE-OFF: COMPLETENESS VS LATENCY
    lateness   0s -> 278/300 counted (92.7%),  22 dropped
    lateness  10s -> 280/300 counted (93.3%),  20 dropped
    lateness  30s -> 282/300 counted (94.0%),  18 dropped
    lateness  60s -> 290/300 counted (96.7%),  10 dropped
    lateness 180s -> 300/300 counted (100.0%),   0 dropped
    Zero lateness closes windows immediately and loses every straggler.
    Large lateness is complete but holds state and delays results.

Look at the first table. By event time every window holds exactly 60 events, because exactly 60 events happened per minute -- the truth. By processing time the counts are 56, 55, 54, 67, 58, and two windows appear after the data ended, containing stragglers. Nothing was lost or duplicated; the arithmetic is correct in both columns. Only one of them answers a question anyone asked.

The second table prices the watermark. At zero lateness you emit immediately and lose 7% of events; at 180 seconds you are complete but every result waits three minutes and the engine holds three minutes of state. Choose by asking what the consumer does with the number: an alerting threshold wants speed, a billing report wants completeness, and a dashboard usually wants speed now plus a corrected value later -- which is exactly what emitting an early result and allowing late updates provides.

Key Takeaways

• Event time is when something happened; processing time is when your code saw it. The gap is variable and unbounded.
• Event-time results are reproducible; processing-time results change on restart and are meaningless for backfills.
• A watermark asserts that no earlier event is still expected, and is normally max event time seen - allowed lateness.
• A window can be emitted and its state freed once the watermark passes its end.
• Allowed lateness trades completeness against latency and state size; there is no setting that optimises all three.
• Measure your actual event-to-arrival delay distribution before choosing lateness; do not guess.
• For dashboards, emit early and allow late corrections rather than choosing one point on the trade-off.

🧪 Practice

1. Compute the distribution of arrival_time - event_time for a real stream you have access to, and report its median and 99th percentile.
2. Rerun the simulation with 40% badly delayed events and report how the completeness-versus-lateness curve changes.
3. Extend the watermark function to route dropped events to a side output, and count them per window.
4. Explain what happens to processing-time windows when you replay three days of history in ten minutes, with a concrete example.
5. Interview: A daily revenue dashboard built on processing-time windows shows a large dip yesterday and a matching spike today, but the finance system shows neither. Explain. (Hint: ask what happened to the pipeline around midnight, and where the missing events ended up rather than whether they were lost.)

Windowing Strategies

An aggregate over an infinite stream is not a number, so before you can compute anything you must cut the stream into finite pieces. The way you cut it is a modelling decision, and three shapes cover nearly every case.

Tumbling windows are fixed-size and non-overlapping. Every event belongs to exactly one window. Use them for periodic reporting: events per minute, revenue per hour. They are the cheapest option, because state for a window is released as soon as it closes.

Sliding (hopping) windows are fixed-size but emitted every slide interval, so they overlap. Use them for smoothed, continuously updated metrics: "average latency over the last 5 minutes, refreshed every 30 seconds". The cost is exactly size / slide: each event belongs to that many windows, multiplying both state and output volume.

Session windows have no fixed size. A window extends while events keep arriving and closes after a gap of inactivity. Use them whenever the unit of analysis is a bout of human activity -- a browsing session, a support conversation, a driving trip. The boundaries come from the data, which is both the point and the difficulty: a session's end is only knowable after waiting out the gap, and one very active user can hold a session open indefinitely.

  THE SAME EVENT TIMELINE UNDER THREE WINDOW TYPES

  events        x x  x x        x           x x x x         x  x
  time     0---------|---------60---------|--------120--------|--------180

  TUMBLING 60   [        w0        ][        w1        ][        w2        ]
                 every event in exactly one window

  SLIDING 60/30 [        w0        ]
                          [        w1        ]
                                    [        w2        ]
                                              [        w3        ]
                 each event in up to size/slide = 2 windows

  SESSION gap 30 [  s1  ]              [   s2   ]        [ s3 ]
                 boundaries set by silence, not by the clock

from collections import defaultdict

# One user's click timestamps, in seconds. Two bursts of activity with a
# long gap between them -- the shape session windows exist to capture.
clicks = [2, 5, 9, 11, 40, 95, 98, 101, 104, 150, 152]

def tumbling(ts, size):
    """Fixed, non-overlapping. Every event lands in exactly one window."""
    w = defaultdict(list)
    for t in ts:
        w[t // size * size].append(t)
    return dict(sorted(w.items()))

def sliding(ts, size, slide):
    """Fixed size, emitted every `slide`. Events land in size/slide windows,
    so state and output volume multiply by that factor."""
    w = defaultdict(list)
    for t in ts:
        first = ((t - size) // slide + 1) * slide
        for start in range(max(first, 0), t + 1, slide):
            if start <= t < start + size:
                w[start].append(t)
    return dict(sorted(w.items()))

def session(ts, gap):
    """Data-driven: a window ends when nothing happens for `gap`. Boundaries
    are determined by the data, not the clock."""
    out, cur = [], [ts[0]]
    for prev, t in zip(ts, ts[1:]):
        if t - prev > gap:
            out.append(cur); cur = [t]
        else:
            cur.append(t)
    out.append(cur)
    return out

print("EVENT TIMESTAMPS:", clicks)

print("\nTUMBLING, size 60")
for start, members in tumbling(clicks, 60).items():
    print(f"  [{start:>3}, {start + 60:>3})  n={len(members):>2}  {members}")

print("\nSLIDING, size 60 slide 30")
sl = sliding(clicks, 60, 30)
for start, members in sl.items():
    print(f"  [{start:>3}, {start + 60:>3})  n={len(members):>2}  {members}")
memberships = sum(len(v) for v in sl.values())
print(f"  {len(clicks)} events produced {memberships} window memberships"
      f" ({memberships / len(clicks):.1f}x state and output)")

print("\nSESSION, inactivity gap 30")
for i, members in enumerate(session(clicks, 30), 1):
    print(f"  session {i}: [{members[0]:>3}, {members[-1]:>3}]"
          f"  n={len(members):>2}  {members}")

  EVENT TIMESTAMPS: [2, 5, 9, 11, 40, 95, 98, 101, 104, 150, 152]

  TUMBLING, size 60
    [  0,  60)  n= 5  [2, 5, 9, 11, 40]
    [ 60, 120)  n= 4  [95, 98, 101, 104]
    [120, 180)  n= 2  [150, 152]

  SLIDING, size 60 slide 30
    [  0,  60)  n= 5  [2, 5, 9, 11, 40]
    [ 30,  90)  n= 1  [40]
    [ 60, 120)  n= 4  [95, 98, 101, 104]
    [ 90, 150)  n= 4  [95, 98, 101, 104]
    [120, 180)  n= 2  [150, 152]
    [150, 210)  n= 2  [150, 152]
    11 events produced 18 window memberships (1.6x state and output)

  SESSION, inactivity gap 30
    session 1: [  2,  40]  n= 5  [2, 5, 9, 11, 40]
    session 2: [ 95, 104]  n= 4  [95, 98, 101, 104]
    session 3: [150, 152]  n= 2  [150, 152]

The same eleven clicks produced three genuinely different descriptions. Tumbling says "5, 4, 2 clicks per minute". Sliding produced 18 memberships from 11 events, and its [30, 90) window contains a single click -- useful for a smoothed curve, misleading if read as a count. Session windows found the structure a human would describe: three bouts of activity, each ending when the user stopped.

	Tumbling	Sliding	Session
Overlap	None	size/slide windows per event	None
Boundaries from	Clock	Clock	Data (gaps)
State cost	Lowest	size/slide x tumbling	Unbounded per active key
Emits	At each boundary	Every slide	After the gap elapses
Natural for	Periodic reports	Moving averages, alerting	User sessions, trips

Two practical warnings. First, sliding windows with a small slide are how streaming jobs quietly become expensive: size 1 hour with a 10-second slide means every event is held in 360 windows. Second, session windows need a timeout as well as a gap, or a user who clicks every 25 seconds for nine hours holds one session -- and its accumulated state -- open all day.

Key Takeaways

• An aggregate over an unbounded stream requires a window; choosing the window type is a modelling decision, not a tuning one.
• Tumbling windows are fixed and non-overlapping, so each event is counted once and state is released promptly.
• Sliding windows overlap, and each event belongs to size/slide of them, multiplying state and output by that factor.
• Session windows are defined by gaps in activity, so boundaries come from the data and are only known after the gap passes.
• A small slide relative to size is the most common cause of accidental state explosion.
• Cap session windows with a maximum duration so one hyperactive key cannot hold state indefinitely.
• Pick the window from the question: counts per period, a smoothed metric, or a bout of human activity.

🧪 Practice

1. Compute tumbling, sliding, and session aggregates over the same click stream and explain in one sentence each what the numbers mean.
2. Calculate state amplification for sliding windows of size 1 hour with slides of 30 minutes, 5 minutes, and 10 seconds.
3. Add a maximum session duration to the session function and show it splitting a nine-hour continuous session.
4. Vary the session gap over 10, 30, and 300 seconds and report how the session count and mean session length change.
5. Interview: Which window type would you use to detect "a user viewed 5 products then abandoned the cart", and what parameters would you need? (Hint: the pattern is defined by what happens between events rather than by any wall-clock interval.)

Message Brokers and Log Systems

A stream processor needs somewhere to read from, and the choice of that "somewhere" determines what your pipeline can do -- particularly whether it can be reprocessed. Two families exist, and they are often confused because both are called queues.

A traditional message queue (RabbitMQ, SQS) delivers a message to a consumer and removes it. It is a transport: once consumed, the message is gone. This suits task dispatch -- send this email, resize this image -- where each message is work to be done once.

A distributed log (Kafka, Kinesis, Pulsar, Redpanda) appends messages to an ordered, durable sequence and retains them. Consumers track their own position by offset, and reading does not consume. This suits data pipelines, because many independent consumers can read the same stream, and any of them can rewind and reprocess. That rewind capability is the single most important architectural property in streaming data: without it, every bug in your streaming logic is permanent.

The log's internal structure is worth knowing precisely, because two hard limits come directly from it:

• A topic is split into partitions, each an independent append-only sequence. A record's key determines its partition, so all records for a key land in one partition.
• Ordering is guaranteed within a partition only. There is no global order across a topic, and there cannot be one without giving up parallelism.
• A consumer group shares the partitions of a topic among its members. Each partition goes to exactly one member, so partition count is the hard ceiling on consumer parallelism -- and it is fixed when the topic is created.

from collections import defaultdict
from zlib import crc32

class PartitionedLog:
    """A Kafka-shaped log: N append-only partitions, each with monotonically
    increasing offsets. Ordering is guaranteed WITHIN a partition only."""
    def __init__(self, n_partitions):
        self.parts = [[] for _ in range(n_partitions)]

    def produce(self, key, value):
        p = crc32(key.encode()) % len(self.parts)   # same key -> same partition
        self.parts[p].append((key, value))
        return p, len(self.parts[p]) - 1            # (partition, offset)

log = PartitionedLog(3)
events = [("order-7", "created"), ("order-9", "created"),
          ("order-7", "paid"), ("order-9", "cancelled"),
          ("order-7", "shipped"), ("order-12", "created"),
          ("order-5", "created"), ("order-5", "paid")]
print("PRODUCING TO A 3-PARTITION LOG")
for k, v in events:
    p, off = log.produce(k, v)
    print(f"  key={k:<9} value={v:<10} -> partition {p}, offset {off}")

print("\nWHAT EACH PARTITION HOLDS")
for i, part in enumerate(log.parts):
    print(f"  p{i}: " + " | ".join(f"{k}:{v}" for k, v in part))
print("  order-7's three events share a partition, so their relative order is")
print("  guaranteed. order-7 versus order-9 has no guaranteed order at all.")

# --- CONSUMER GROUPS: PARALLELISM IS CAPPED BY PARTITION COUNT. ---
def assign(n_partitions, n_consumers):
    a = defaultdict(list)
    for p in range(n_partitions):
        a[p % n_consumers].append(p)            # round-robin assignment
    return dict(a)

print("\nCONSUMER GROUP ASSIGNMENT FOR 3 PARTITIONS")
for n in (1, 2, 3, 5):
    a = assign(3, n)
    idle = n - len([c for c in a if a[c]])
    desc = ", ".join(f"c{c}->{a.get(c, [])}" for c in range(n))
    print(f"  {n} consumers: {desc}   idle: {idle}")
print("  A fourth consumer cannot help: partition count is the hard ceiling")
print("  on consumer parallelism, and it is chosen at topic creation.")

# --- RETENTION VS COMPACTION. ---
print("\nTWO WAYS TO BOUND A LOG'S SIZE")
raw = [("u1", "v1"), ("u2", "v1"), ("u1", "v2"), ("u3", "v1"), ("u1", "v3")]
fmt = lambda pairs: " ".join(f"{k}={v}" for k, v in pairs)
print(f"  raw log, {len(raw)} records      {fmt(raw)}")
print(f"  time retention, keep last 3  {fmt(raw[-3:])}")
compacted = list({k: v for k, v in raw}.items())     # last value per key wins
print(f"  log compaction, last per key {fmt(compacted)}")
print("  Retention suits event streams; compaction turns a log into a")
print("  replayable snapshot of current state, which is what a changelog is.")

  PRODUCING TO A 3-PARTITION LOG
    key=order-7   value=created    -> partition 2, offset 0
    key=order-9   value=created    -> partition 0, offset 0
    key=order-7   value=paid       -> partition 2, offset 1
    key=order-9   value=cancelled  -> partition 0, offset 1
    key=order-7   value=shipped    -> partition 2, offset 2
    key=order-12  value=created    -> partition 0, offset 2
    key=order-5   value=created    -> partition 1, offset 0
    key=order-5   value=paid       -> partition 1, offset 1

  WHAT EACH PARTITION HOLDS
    p0: order-9:created | order-9:cancelled | order-12:created
    p1: order-5:created | order-5:paid
    p2: order-7:created | order-7:paid | order-7:shipped
    order-7's three events share a partition, so their relative order is
    guaranteed. order-7 versus order-9 has no guaranteed order at all.

  CONSUMER GROUP ASSIGNMENT FOR 3 PARTITIONS
    1 consumers: c0->[0, 1, 2]   idle: 0
    2 consumers: c0->[0, 2], c1->[1]   idle: 0
    3 consumers: c0->[0], c1->[1], c2->[2]   idle: 0
    5 consumers: c0->[0], c1->[1], c2->[2], c3->[], c4->[]   idle: 2
    A fourth consumer cannot help: partition count is the hard ceiling
    on consumer parallelism, and it is chosen at topic creation.

  TWO WAYS TO BOUND A LOG'S SIZE
    raw log, 5 records      u1=v1 u2=v1 u1=v2 u3=v1 u1=v3
    time retention, keep last 3  u1=v2 u3=v1 u1=v3
    log compaction, last per key u1=v3 u2=v1 u3=v1
    Retention suits event streams; compaction turns a log into a
    replayable snapshot of current state, which is what a changelog is.

The two idle consumers are the lesson most teams learn the expensive way. Consumer lag was growing, so they scaled the consumer group from three to five -- and throughput did not change, because there were only three partitions to assign. Repartitioning a live topic is disruptive, so partition count deserves deliberate thought at creation: enough for peak parallelism plus growth headroom, but not so many that per-partition overhead and file handles become the problem.

The retention comparison shows the other design choice. Time or size retention keeps recent records and discards old ones, which fits event streams. Log compaction keeps the latest record per key forever, which turns the log into a replayable snapshot of current state -- the mechanism behind change-data-capture streams and Kafka Streams' state stores. If your topic represents "the current state of each entity", you want compaction; if it represents "things that happened", you want retention.

Key Takeaways

• A queue delivers and deletes; a log appends and retains, letting many consumers read and rewind independently.
• Replayability is the decisive property: without it, a bug in streaming logic cannot be corrected retroactively.
• Topics are split into partitions; a record's key selects its partition.
• Ordering holds within a partition only, so pick keys that put causally related records together.
• Partition count caps consumer parallelism and is fixed at topic creation -- adding consumers beyond it does nothing.
• Retention bounds a log by time or size; compaction keeps the latest record per key, making the log a state snapshot.
• Consumers own their offsets, which is why offset commit timing determines delivery semantics.

🧪 Practice

1. Simulate a 4-partition log and verify that all records for one key land in one partition while two keys may not preserve relative order.
2. Compute consumer group assignments for 12 partitions and 1, 5, 12, and 20 consumers, and identify where added consumers stop helping.
3. Implement log compaction over a stream of keyed updates and show the result is a snapshot of current state.
4. Choose a partition key for an order-events topic such that all events for an order stay ordered, and state what skew risk your choice carries.
5. Interview: Consumer lag is growing on a 6-partition topic and you already run 6 consumers. What are your options? (Hint: one option changes the topic, one changes what each consumer does per record, and one changes where the slow part actually is.)

Real-Time Feature Computation

This is where streaming meets machine learning, and it contains the most expensive bug in applied data science.

A model in production needs features computed from recent behaviour: how many transactions has this card seen in the last ten minutes, how many pages has this session touched, what is the user's average order value this week. Two requirements collide.

1. At serving time, the feature must be available in single-digit milliseconds, computed from data that may be seconds old.
2. At training time, the same feature must be computed for millions of historical rows, each as of that row's timestamp.

The trap is that requirement 2 is easy to get wrong in a way that looks correct. If you compute "transactions in the last ten minutes" with a GROUP BY over your full history table, you are using data from after each training row's decision point. The model learns from the future, scores brilliantly offline, and degrades in production. This is label leakage via feature computation, and the discipline that prevents it is the point-in-time correct join: for each training row, aggregate only records whose event time is strictly before that row's timestamp.

  POINT-IN-TIME CORRECTNESS

  transactions   t1   t2        t3        t4   t5
                 |    |         |         |    |
  ---------------+----+---------+----+----+----+--------> event time
                                     ^
                              decision at T

  CORRECT   feature at T = f(t1, t2, t3)        only the past
  LEAKY     feature at T = f(t1..t5)            includes t4, t5

  The leaky feature cannot be reproduced at serving time, because at
  time T the events t4 and t5 have not happened yet.

The architecture that resolves both requirements is the feature store, which is best understood as two synchronised views of one definition:

• The offline store (a warehouse or lakehouse table) holds full history and answers point-in-time queries for training.
• The online store (Redis, DynamoDB, Cassandra) holds only the current feature value per entity and answers point lookups for serving.
• Both are written from the same definition, so the numbers agree. If training reads a Spark job's output and serving reads a hand-written service, they will diverge, and that divergence is train-serve skew.

import random
from bisect import bisect_right
from collections import deque, defaultdict

rng = random.Random(5)

# --- THE DATA: TRANSACTIONS, AND LABELLED DECISIONS MADE AT SOME TIMES. ---
# A fraud model wants "how much has this user spent in the last 600s"
# AS OF the decision time.
txs = sorted((f"u{rng.randrange(20)}", t, round(rng.expovariate(1 / 50), 2))
             for t in rng.sample(range(0, 5000), 1200))
decisions = sorted((f"u{rng.randrange(20)}", t)
                   for t in rng.sample(range(1000, 5000), 300))

by_user = defaultdict(list)
for u, t, a in txs:
    by_user[u].append((t, a))

WINDOW = 600

def feature_correct(user, as_of):
    """POINT-IN-TIME CORRECT: only transactions strictly before as_of."""
    hist = by_user[user]
    times = [t for t, _ in hist]
    hi = bisect_right(times, as_of - 1)                 # exclude as_of itself
    return sum(a for t, a in hist[:hi] if t > as_of - WINDOW)

def feature_leaky(user, as_of):
    """WRONG: computed from the whole table, so it includes the future. This
    is what a naive window over all history produces."""
    hist = by_user[user]
    return sum(a for t, a in hist if abs(t - as_of) < WINDOW)

rows = [(u, t, feature_correct(u, t), feature_leaky(u, t)) for u, t in decisions]
n_diff = sum(1 for _, _, c, l in rows if abs(c - l) > 1e-9)
mean_c = sum(c for _, _, c, _ in rows) / len(rows)
mean_l = sum(l for _, _, _, l in rows) / len(rows)

print(f"TRAINING ROWS: {len(rows)} decisions over {len(txs)} transactions")
print(f"  rows where the two differ  : {n_diff} ({n_diff / len(rows):.0%})")
print(f"  mean point-in-time feature : {mean_c:8.2f}")
print(f"  mean leaky feature         : {mean_l:8.2f}"
      f"  ({mean_l / mean_c - 1:+.0%})")
print("  The leaky version is systematically larger because it sums a window")
print("  centred on the decision instead of ending at it. A model trained on")
print("  it learns from the future and cannot reproduce that when serving.")
print("\n  SAMPLE ROWS")
for u, t, c, l in rows[:5]:
    flag = "  <-- leaked" if abs(c - l) > 1e-9 else ""
    print(f"    {u:>4} at t={t:<5} correct={c:>8.2f}  leaky={l:>8.2f}{flag}")

# --- THE STREAMING VERSION: ONE PASS, BOUNDED STATE PER KEY. ---
class SlidingSum:
    """What an online feature service actually runs: a per-key deque holding
    only the events still inside the window. Bounded memory per key, O(1)
    amortised per event -- no scan of history at request time."""
    def __init__(self, window):
        self.window, self.q, self.total = window, deque(), 0.0

    def add(self, t, a):
        self.q.append((t, a)); self.total += a
        self._evict(t)

    def value(self, now):
        self._evict(now)
        return round(self.total, 2)

    def _evict(self, now):
        while self.q and self.q[0][0] <= now - self.window:
            self.total -= self.q.popleft()[1]

state = defaultdict(lambda: SlidingSum(WINDOW))
online = {}
merged = sorted([(t, "tx", u, a) for u, t, a in txs]
                + [(t, "ask", u, None) for u, t in decisions])
for t, kind, u, a in merged:                  # single chronological pass
    if kind == "tx":
        state[u].add(t, a)
    else:
        online[(u, t)] = state[u].value(t)

agree = sum(1 for (u, t), v in online.items()
            if abs(v - round(feature_correct(u, t), 2)) < 0.011)
print(f"\nSTREAMING FEATURE MATCHES THE BATCH FEATURE:"
      f" {agree}/{len(online)} rows")
print(f"  max deque length held: {max(len(s.q) for s in state.values())}"
      f" events, for {len(state)} keys")
print("  Same definition, two implementations, one number. That is the whole")
print("  job of a feature store.")

  TRAINING ROWS: 300 decisions over 1200 transactions
    rows where the two differ  : 294 (98%)
    mean point-in-time feature :   366.72
    mean leaky feature         :   692.87  (+89%)
    The leaky version is systematically larger because it sums a window
    centred on the decision instead of ending at it. A model trained on
    it learns from the future and cannot reproduce that when serving.

    SAMPLE ROWS
      u0 at t=1111  correct=  363.64  leaky=  548.77  <-- leaked
      u0 at t=1126  correct=  363.64  leaky=  548.77  <-- leaked
      u0 at t=1331  correct=  450.66  leaky=  671.82  <-- leaked
      u0 at t=1371  correct=  450.66  leaky=  794.14  <-- leaked
      u0 at t=1711  correct=  185.13  leaky=  522.64  <-- leaked

  STREAMING FEATURE MATCHES THE BATCH FEATURE: 300/300 rows
    max deque length held: 11 events, for 20 keys
    Same definition, two implementations, one number. That is the whole
    job of a feature store.

Ninety-eight percent of training rows carried a different feature value, with the leaky version 89% larger on average. A model trained on that column would report excellent offline metrics and then fail, and the failure would look like drift rather than like a bug -- which is why this specific error survives so long in so many projects.

The second half shows the resolution. One chronological pass, a bounded deque per key, and the streaming feature agreed with the point-in-time batch feature on all 300 rows. The state held was 11 events per key, not the full history, which is what makes this viable at millions of keys.

Practical guidance that follows:

• Prefer aggregations with bounded incremental state: counts, sums, means via (sum, count), min and max over a window, and sketches (HyperLogLog for distinct counts, t-digest for quantiles). Avoid exact medians and exact distinct counts over long windows.
• Log the features that were actually served, with the prediction. It is the only ground truth about skew, and it lets you build training data from what production really saw.
• Reconcile online and offline continuously: sample entities, compute the feature both ways, alert on divergence. Skew appears silently after code changes on either side.
• Version feature definitions. Changing a window from 10 to 15 minutes is a new feature, not an edit, because every historical value changes with it.

Key Takeaways

• Serving needs millisecond feature lookups; training needs the same feature computed as of each historical row's timestamp.
• Aggregating over a full history table for training leaks future data and is the most common cause of offline-brilliant, online-useless models.
• A point-in-time correct join uses only records strictly before each row's decision time.
• A feature store keeps an offline store for training history and an online store for current values, written from one shared definition.
• Incremental, bounded-state aggregations (sums, counts, sketches) are what can be maintained online; exact medians and distinct counts cannot.
• Log served features with predictions; it is the only reliable evidence about train-serve skew.
• Treat a changed window or definition as a new feature version, since every historical value changes.

🧪 Practice

1. Compute a "count in last 24 hours" feature for a set of labelled events twice -- naively and point-in-time correct -- and report the share of rows that differ.
2. Implement a bounded sliding-window sum with a deque and verify it matches a full recomputation at every request time.
3. Replace an exact distinct count in a streaming feature with a HyperLogLog sketch and compare accuracy against state size.
4. Design a reconciliation job that samples 1,000 entities hourly and alerts when online and offline feature values diverge by more than a threshold.
5. Interview: A fraud model scores AUC 0.95 offline and 0.71 in production. Name the first thing you would check and how. (Hint: compare the feature values logged at serving time with the feature values in the training table for the same entities and timestamps.)

<a id="144-performance-engineering"></a>

14.4 Performance Engineering

A distributed job that takes six hours is rarely six hours of useful work, and the difference is almost always recoverable once you know where to look; this section covers how to find the actual bottleneck instead of guessing, how to keep large data inside the memory you have, when caching helps and when it quietly costs you, and how to reason about the bill that arrives at the end of the month.

Profiling Data Workloads

The first rule of performance work is that intuition about where time goes is usually wrong, and the second is that it costs almost nothing to check. Before changing any code, identify which of four resources is the constraint, because each has a different fix and the fixes do not substitute for one another:

Bottleneck	Symptom	Typical fix
CPU	Cores pinned, low I/O wait	Vectorise, reduce per-row work, better algorithm
I/O	Low CPU, high disk or network read	Columnar formats, projection, compression, pruning
Memory	Swapping, GC thrash, spills to disk, OOM kills	Smaller types, chunking, fewer cached datasets
Network / shuffle	Long stages with large shuffle read	Broadcast joins, pre-aggregation, fewer partitions

The tools, from coarse to fine:

• Wall-clock timing around blocks. Crude and often sufficient: it tells you which of five pipeline steps owns the runtime.
• cProfile for per-function cumulative time in a single process. Sort by cumulative time to find the expensive subtree, by total time to find the expensive function itself.
• line_profiler when a function is known to be slow but not why.
• memory_profiler / tracemalloc for allocation growth.
• The Spark UI for distributed jobs, where the interesting quantities are per-stage: task duration distribution (max versus median exposes skew), shuffle read and write bytes, spill to disk, and GC time.

A per-stage habit worth building in Spark: compare the maximum task duration with the median. If the maximum is many times the median, you have skew and no amount of extra executors will help. If all tasks are uniformly slow, you have a genuine compute or I/O problem.

import cProfile, pstats, time
import numpy as np, pandas as pd

rng = np.random.default_rng(0)
N = 300_000
df = pd.DataFrame({
    "user": rng.integers(0, 50_000, N),
    "country": rng.choice(["IT", "DE", "FR", "ES"], N),
    "amount": rng.exponential(40, N).round(2),
})

# --- THE SUSPECT: A ROW-BY-ROW LOOP HIDDEN BEHIND A TIDY API. ---
def score_rowwise(df):
    out = []
    for _, row in df.iterrows():          # one Series object per row
        out.append(row["amount"] * (1.2 if row["country"] == "IT" else 1.0))
    return pd.Series(out, index=df.index)

def score_vectorised(df):
    mult = np.where(df["country"].values == "IT", 1.2, 1.0)
    return pd.Series(df["amount"].values * mult, index=df.index)

def timed(label, fn, *a):
    t = time.perf_counter(); out = fn(*a); dt = time.perf_counter() - t
    print(f"  {label:26s} {dt:7.3f}s")
    return out, dt

print(f"SCORING {N:,} ROWS TWO WAYS")
a, t_slow = timed("iterrows loop", score_rowwise, df)
b, t_fast = timed("vectorised numpy", score_vectorised, df)
assert np.allclose(a.values, b.values)          # same answer, both ways
print(f"  identical results, {t_slow / t_fast:,.0f}x apart")

# --- PROFILE IT: THE POINT IS WHICH LINE, NOT THE TOTAL. ---
print("\nWHERE THE SLOW VERSION SPENDS ITS TIME (60,000 rows)")
pr = cProfile.Profile(); pr.enable()
score_rowwise(df.head(60_000))
pr.disable()
st = pstats.Stats(pr).sort_stats("cumulative")
rows = sorted(st.stats.items(), key=lambda kv: -kv[1][3])[:6]
print(f"  {'ncalls':>8} {'tottime':>8} {'cumtime':>8}  where")
for (fn, lineno, name), (cc, nc, tt, ct, _) in rows:
    where = f"{fn.rsplit('/', 1)[-1]}:{lineno}({name})"
    print(f"  {nc:>8} {tt:>8.3f} {ct:>8.3f}  {where[:44]}")
print("  60,000 iterations produced ~8.7 million function calls: pandas built")
print("  a Series object per row, and the arithmetic is not the cost at all.")

  SCORING 300,000 ROWS TWO WAYS
    iterrows loop                9.374s
    vectorised numpy             0.010s
    identical results, 937x apart

  WHERE THE SLOW VERSION SPENDS ITS TIME (60,000 rows)
      ncalls  tottime  cumtime  where
           1    0.192    6.533  profiling.py:13(score_rowwise)
       60001    0.200    5.219  frame.py:1538(iterrows)
       60001    0.579    4.576  series.py:369(__init__)
       60001    0.746    1.615  construction.py:531(sanitize_array)
      120000    0.233    1.099  series.py:945(__getitem__)
       60001    0.181    0.743  managers.py:2053(from_array)
    60,000 iterations produced ~8.7 million function calls: pandas built
    a Series object per row, and the arithmetic is not the cost at all.

The profile says something a timer could not. The expensive frames are Series.__init__, sanitize_array, and from_array -- none of which appear in the code. The multiplication is nowhere in the top six. The cost is object construction per row, which is why the fix is not a faster multiplication but the removal of the loop. Three orders of magnitude, from reading a profile rather than guessing.

A disciplined loop for this work:

1. Measure the whole pipeline and find which step owns the time.
2. Classify that step: CPU, I/O, memory, or network.
3. Profile inside it to find the specific frame or stage.
4. Change one thing.
5. Re-measure, and keep the change only if the number moved.

Step 5 is skipped more often than any other, and it is the one that prevents a codebase from accumulating "optimisations" that cost readability and buy nothing.

Key Takeaways

• Identify which resource is the constraint -- CPU, I/O, memory, or network -- before changing anything, because the fixes do not overlap.
• Time coarse blocks first, then profile inside the step that owns the runtime.
• Sort profiles by cumulative time to find the expensive subtree and by total time to find the expensive function.
• In Spark, compare maximum to median task duration per stage: a large ratio means skew, uniform slowness means real work.
• Watch shuffle read/write bytes, disk spill, and GC time per stage; they name the bottleneck directly.
• Row-wise iteration over a DataFrame costs object construction per row and is routinely 100-1000x slower than the vectorised equivalent.
• Always re-measure after a change, and revert anything that did not move the number.

🧪 Practice

1. Profile a pipeline of yours with cProfile and report which function holds the largest cumulative time versus the largest total time.
2. Replace one iterrows or apply(axis=1) loop with a vectorised expression and measure the speedup.
3. For a Spark stage you have run, record median and maximum task duration and decide from the ratio whether skew is present.
4. Instrument a five-step pipeline with timers and produce a table of time per step and its percentage of total.
5. Interview: A nightly job's runtime grew from 40 minutes to 4 hours with no code change. What do you check, in what order? (Hint: something other than the code changed -- consider data volume, data distribution, and resources the cluster shares with others.)

Memory Management for Large Datasets

Out-of-memory failures feel like a hardware limit, and they usually are not. Most large datasets are stored in representations several times larger than necessary, and most computations hold more of them at once than they need.

Four levers, in the order they pay off.

1. Fit the type to the data. A column of five country codes does not need variable-length strings; a user id below 100,000 does not need 64 bits; a currency amount does not need double precision. category dtype replaces repeated values with small integer codes plus one dictionary. On older pandas versions, where string columns were Python objects, this alone often cut memory by 10x; from pandas 3 strings are Arrow-backed and already compact, so the win is smaller but the principle is unchanged.

2. Project early. Reading 60 columns to use 3 wastes 95% of the memory and all of the I/O. usecols at read time, columns= in Parquet readers, and SELECT lists in SQL are the cheapest optimisation available anywhere.

3. Chunk. Process a fixed number of rows at a time and keep only the aggregate. Peak memory then depends on chunk size, not file size, and the same code handles 1 GB and 1 TB.

4. Go out-of-core or columnar-native. Let a library that was designed for this -- DuckDB, Polars in streaming mode, Dask, Arrow datasets -- manage spilling and batching for you.

  WHY NUMBERS BEAT OBJECTS

  object / variable-width strings      category (dictionary encoding)
    row0 -> "IT"  (separate buffer)      codes:  [0,1,0,2,1,0, ...]  int8
    row1 -> "DE"  (separate buffer)      dict:   {0:"IT", 1:"DE", 2:"FR"}
    row2 -> "IT"  (another buffer)
    ...                                  1 byte per row + a tiny mapping,
    bytes per row grows with the          independent of string length
    string and its bookkeeping

import os, tempfile
import numpy as np, pandas as pd

rng = np.random.default_rng(0)
N = 2_000_000
raw = pd.DataFrame({
    "user_id": rng.integers(0, 60_000, N),                       # int64
    "country": rng.choice(["IT", "DE", "FR", "ES", "PL"], N),    # strings
    "device":  rng.choice(["ios", "android", "web"], N),
    "amount":  rng.exponential(40, N),                           # float64
    "is_new":  rng.integers(0, 2, N).astype(bool),
})

def mb(df): return df.memory_usage(deep=True).sum() / 1024 ** 2

print(f"{N:,} ROWS, DEFAULT DTYPES")
print(raw.dtypes.to_string())
print(f"  total {mb(raw):8.1f} MB")
print("\n  per column")
for c in raw.columns:
    print(f"    {c:9s} {raw[c].dtype!s:8s}"
          f" {raw[c].memory_usage(deep=True) / 1024 ** 2:8.1f} MB")

# --- FIT THE TYPE TO THE DATA, NOT THE OTHER WAY ROUND. ---
tight = raw.astype({
    "user_id": "int32",     # 60,000 values fit comfortably in 32 bits
    "country": "category",  # 5 distinct values -> int8 codes + a tiny mapping
    "device": "category",   # 3 distinct values
    "amount": "float32",    # 7 significant digits is plenty for currency
})
print(f"\nAFTER TYPE TIGHTENING: {mb(tight):.1f} MB"
      f"  ({mb(raw) / mb(tight):.1f}x smaller)")
for c in tight.columns:
    print(f"    {c:9s} {tight[c].dtype!s:9s}"
          f" {tight[c].memory_usage(deep=True) / 1024 ** 2:8.1f} MB")
assert (tight["country"] == raw["country"]).all()        # same values
print("  Values are unchanged; only their representation is.")

# --- CHUNKED PROCESSING: CONSTANT MEMORY OVER AN ARBITRARILY LARGE FILE. ---
d = tempfile.mkdtemp(); path = os.path.join(d, "big.csv")
raw.to_csv(path, index=False)
file_mb = os.path.getsize(path) / 1024 ** 2

total, count, peak_chunk_mb = 0.0, 0, 0.0
for chunk in pd.read_csv(path, chunksize=200_000,
                         usecols=["country", "amount"],   # project at read time
                         dtype={"country": "category"}):
    peak_chunk_mb = max(peak_chunk_mb, mb(chunk))
    total += chunk["amount"].sum()
    count += len(chunk)
print(f"\nSTREAMING A {file_mb:.0f} MB CSV IN 200,000-ROW CHUNKS")
print(f"  rows {count:,}  mean amount {total / count:.4f}")
print(f"  largest chunk resident at once: {peak_chunk_mb:.1f} MB")
print(f"  peak memory is set by chunk size, not by file size")

  2,000,000 ROWS, DEFAULT DTYPES
  user_id      int64
  country        str
  device         str
  amount     float64
  is_new        bool
    total     75.0 MB

    per column
      user_id   int64        15.3 MB
      country   str          19.1 MB
      device    str          23.5 MB
      amount    float64      15.3 MB
      is_new    bool          1.9 MB

  AFTER TYPE TIGHTENING: 21.0 MB  (3.6x smaller)
      user_id   int32          7.6 MB
      country   category       1.9 MB
      device    category       1.9 MB
      amount    float32        7.6 MB
      is_new    bool           1.9 MB
    Values are unchanged; only their representation is.

  STREAMING A 73 MB CSV IN 200,000-ROW CHUNKS
    rows 2,000,000  mean amount 39.9492
    largest chunk resident at once: 1.7 MB
    peak memory is set by chunk size, not by file size

Three numbers carry the topic. The same 2 million rows occupy 75 MB or 21 MB depending only on declared types. The device column -- three distinct values -- was the single largest consumer at 23.5 MB and became 1.9 MB as a category. And a 73 MB file was aggregated with 1.7 MB resident, because projection plus chunking makes peak memory independent of file size.

In the JVM world the same reasoning applies with different vocabulary. Executor memory is divided between execution (shuffle buffers, sorts, joins) and storage (cached datasets), and they compete. Cache too much and shuffles spill to disk. Symptoms to recognise:

Symptom	Likely cause
Spill to disk in a stage	Partitions too large, or storage starving execution
High GC time	Too many small objects; often RDDs or Python UDFs
Container killed by the OS	Off-heap or Python worker memory not budgeted
Driver OOM	collect() on a large result, or a huge broadcast

Key Takeaways

• Most memory pressure is representation waste, not a genuine hardware limit.
• Declare the smallest correct dtype: int32, float32, and category for low-cardinality strings.
• A low-cardinality string column is often the largest one in a frame, and dictionary encoding makes it among the smallest.
• Project columns at read time; reading what you never use costs memory and I/O for no benefit.
• Chunking makes peak memory a function of chunk size rather than dataset size, and the same code then scales.
• In Spark, cached storage and execution memory compete; over-caching causes shuffle spills.
• Read the failure mode: disk spill, GC time, container kill, and driver OOM each point at a different cause.

🧪 Practice

1. Load a dataset with default dtypes, report its memory, then tighten every column and report the ratio.
2. Convert your highest-cardinality string column to category and explain why the saving is smaller than for a low-cardinality one.
3. Compute a group-wise mean over a file too large for memory using chunks, and verify it matches the full-file result.
4. Trigger a deliberate disk spill in Spark by lowering executor memory, then find the spill metric in the stage detail.
5. Interview: A pandas job fails with an OOM error on a 20 GB machine while reading an 8 GB CSV. Explain how that is possible. (Hint: think about what the in-memory representation costs relative to the file, and how many copies exist during parsing.)

Caching and Materialization

Caching trades memory for repeated computation, and like every trade it can be made at a loss. The decision rule is short: cache what is expensive to produce and will be read more than once. Everything else is overhead.

In Spark, cache() and persist() mark a dataset for materialisation on its next action. The storage level chooses what you pay:

Level	Stored as	Notes
MEMORY_ONLY	Deserialised objects in RAM	Fastest read; partitions that do not fit are recomputed
MEMORY_AND_DISK	RAM, overflow to disk	The sensible default; cache() on a DataFrame means this
MEMORY_ONLY_SER	Serialised bytes in RAM	Smaller, costs CPU per read
DISK_ONLY	Local disk	For when recomputation is worse than disk I/O
*_2 variants	Two copies on different nodes	Faster recovery, double the space

Two facts about Spark caching are routinely missed, and both appear in the measurement below. First, caching is lazy: marking a DataFrame does nothing until an action materialises it, so the first action after cache() is not faster. Second, Spark already reuses shuffle output files between jobs, so a post-shuffle result is cheap to re-read without any cache -- the datasets worth caching are the ones that are expensive and narrow.

import time
from pyspark.sql import SparkSession, functions as F
from pyspark import StorageLevel

spark = (SparkSession.builder.master("local[4]").appName("caching")
         .config("spark.sql.shuffle.partitions", 8).getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

def build():
    """An expensive NARROW pipeline: no shuffle, so there are no shuffle
    files for Spark to reuse implicitly. Every action recomputes all of it."""
    df = spark.range(4_000_000)
    x = F.col("id").cast("double") + 1
    for _ in range(12):                       # deliberately heavy arithmetic
        x = F.sqrt(F.abs(F.sin(x) * 1000) + 1)
    return df.withColumn("v", x).filter(F.col("v") > 0)

def three_actions(df):
    """Three DIFFERENT actions -- the realistic pattern in exploratory work."""
    t = time.perf_counter()
    df.count()
    df.agg(F.sum("v")).collect()
    df.agg(F.max("v")).collect()
    return time.perf_counter() - t

print("THREE DIFFERENT ACTIONS ON THE SAME NARROW PIPELINE")
uncached = build()
t_un = three_actions(uncached)
print(f"  no cache: computed 3 times        {t_un:6.2f}s")

cached = build().cache()
cached.count()                                # first action materialises it
t_c = three_actions(cached)
print(f"  cached:   computed once, read 3x  {t_c:6.2f}s"
      f"   ({t_un / t_c:.1f}x faster)")
cached.unpersist()

print("\nCACHING SOMETHING USED ONLY ONCE")
once_plain = build()
t = time.perf_counter(); once_plain.agg(F.sum("v")).collect()
plain = time.perf_counter() - t
once_cached = build().persist(StorageLevel.MEMORY_AND_DISK)
t = time.perf_counter(); once_cached.agg(F.sum("v")).collect()
with_cache = time.perf_counter() - t
print(f"  single action, no cache    {plain:5.2f}s")
print(f"  single action, persisted   {with_cache:5.2f}s"
      f"   ({with_cache / plain:.1f}x the cost)")
print("  A cache read once is pure overhead: you pay to materialise it, never")
print("  collect the benefit, and take memory away from execution.")
once_cached.unpersist()

print("\nWHAT SPARK REUSES WITHOUT BEING ASKED")
shuffled = (spark.range(4_000_000)
            .withColumn("k", F.col("id") % 100)
            .groupBy("k").count())
t = time.perf_counter(); shuffled.collect(); first = time.perf_counter() - t
t = time.perf_counter(); shuffled.collect(); second = time.perf_counter() - t
print(f"  post-shuffle aggregate, 1st action {first:5.2f}s")
print(f"  post-shuffle aggregate, 2nd action {second:5.2f}s")
print("  Shuffle output files survive between jobs, so a post-shuffle result")
print("  is already cheap to re-read. Cache what is EXPENSIVE AND NARROW.")
spark.stop()

  THREE DIFFERENT ACTIONS ON THE SAME NARROW PIPELINE
    no cache: computed 3 times          6.64s
    cached:   computed once, read 3x    0.98s   (6.8x faster)

  CACHING SOMETHING USED ONLY ONCE
    single action, no cache     1.33s
    single action, persisted    1.82s   (1.4x the cost)
    A cache read once is pure overhead: you pay to materialise it, never
    collect the benefit, and take memory away from execution.

  WHAT SPARK REUSES WITHOUT BEING ASKED
    post-shuffle aggregate, 1st action  0.66s
    post-shuffle aggregate, 2nd action  0.06s
    Shuffle output files survive between jobs, so a post-shuffle result
    is already cheap to re-read. Cache what is EXPENSIVE AND NARROW.

Three results, three rules. Reused expensive work: 6.8x faster cached. Single-use work: 1.4x slower cached, because materialisation was paid for and never amortised. And a post-shuffle aggregate went from 0.66s to 0.06s on its second action with no cache at all, because the shuffle files were still there -- so caching it would have bought nothing while consuming memory.

Beyond in-job caching, materialisation is the same idea at pipeline scale: persist an intermediate result as a table so downstream jobs read instead of recompute.

• A materialised view or summary table replaces a query that many dashboards run repeatedly. The trade is freshness against cost.
• Incremental materialisation processes only new partitions rather than rebuilding history -- the difference between a 3-minute and a 3-hour nightly job.
• Medallion layering (raw, cleaned, aggregated) is materialisation applied as architecture: each layer is computed once and read by many consumers.

The costs are real and belong in the decision: storage, a pipeline to keep it fresh, staleness for consumers, and another thing to backfill when a definition changes. Materialise a query when its cost times its frequency exceeds the cost of maintaining the table -- and when nobody needs an answer fresher than your refresh interval.

Key Takeaways

• Cache only what is expensive to compute and read more than once; everything else is overhead.
• cache() is lazy -- the first action after it pays full price.
• MEMORY_AND_DISK is the sensible default; MEMORY_ONLY silently recomputes partitions that do not fit.
• Spark reuses shuffle files across jobs, so post-shuffle results are already cheap; cache expensive narrow pipelines instead.
• Caching a single-use dataset measurably slows it down and steals memory from execution.
• Call unpersist() when done; cached data competes with shuffles for the same budget.
• Materialised tables are caching at pipeline scale: worthwhile when query cost times frequency beats maintenance cost, and staleness is acceptable.
• Prefer incremental materialisation over full rebuilds.

🧪 Practice

1. Build an expensive narrow pipeline, run three different actions with and without cache(), and report both timings.
2. Compare MEMORY_ONLY against MEMORY_AND_DISK on a dataset larger than the executor memory and explain the difference.
3. Find a query your team runs repeatedly, estimate its monthly cost, and compare against the cost of a materialised summary table.
4. Convert a full-rebuild aggregation into one that processes only new partitions, and measure both.
5. Interview: A colleague caches every DataFrame in a 12-step pipeline "to be safe". What problems would you expect? (Hint: consider what cached storage competes with inside an executor, and how many of those twelve datasets are read more than once.)

Cost-Aware Cloud Computing

On-premise hardware made performance work optional: the machines were already paid for, so a slow job was merely annoying. In the cloud, compute is metered, and a badly written query has a price you can look up. Cost becomes an engineering property of your code, and often the one leadership notices first.

The four billing dimensions that matter:

1. Compute time, usually per vCPU-second. Per-second billing means total vCPU-hours matter more than cluster shape.
2. Storage, per GB-month, with tiers that trade holding cost against retrieval cost.
3. Data scanned, on serverless warehouses -- billed per TB read by a query, which is why pruning is a financial control.
4. Data transfer, especially egress across regions or to the internet. Almost always the surprise line on a first bill.

Then there are the purchasing options, which are where the large savings live: spot or preemptible instances at 60-90% off but interruptible; reserved or committed use at 30-60% off for a 1-3 year commitment; autoscaling and auto-termination so idle clusters stop costing money; and serverless, where you pay per query and nothing between them.

# Illustrative list prices (USD, eu-west-1 class). Substitute your own; the
# STRUCTURE of the arithmetic is what transfers, not the numbers.
ON_DEMAND_PER_VCPU_HR = 0.0425
SPOT_DISCOUNT = 0.70                  # 70% off, interruptible
STORAGE_PER_GB_MONTH = {"standard": 0.023, "infrequent": 0.0125,
                        "archive": 0.004}
SCAN_PER_TB = 5.00                    # serverless warehouse, per TB scanned
EGRESS_PER_GB = 0.09                  # cross-region / internet
RETRIEVAL_PER_GB = {"standard": 0.0, "infrequent": 0.01, "archive": 0.03}

def cluster_cost(vcpus, hours, spot=False):
    rate = ON_DEMAND_PER_VCPU_HR * (1 - SPOT_DISCOUNT if spot else 1)
    return vcpus * hours * rate

print("THE SAME WORK, DIFFERENT SHAPES (48 vCPU-hours of compute)")
for vcpus, hours in [(8, 6.0), (48, 1.0), (192, 0.25)]:
    c = cluster_cost(vcpus, hours)
    print(f"  {vcpus:>3} vCPUs x {hours:>4.2f} h = ${c:6.2f}"
          f"   wall clock {hours * 60:>5.0f} min")
print("  Per-second billing makes these nearly equal in cost, so buy the")
print("  wall-clock time you need -- as long as the work really scales.")

print("\nAMDAHL MEETS THE INVOICE (90% parallel job, 1 vCPU = 10 h)")
for n in (1, 4, 16, 64, 256):
    speedup = 1 / (0.10 + 0.90 / n)          # the serial 10% never shrinks
    hours = 10 / speedup
    print(f"  {n:>4} vCPUs  wall {hours:6.2f} h  cost"
          f" ${cluster_cost(n, hours):7.2f}  efficiency {speedup / n:6.1%}")
print("  Scaling out buys latency, and past a point you pay 20x the money")
print("  for 2x the speed. That can be the right call -- but name it.")

print("\nSPOT VS ON-DEMAND FOR A NIGHTLY 40-vCPU, 3-HOUR ETL JOB")
od = cluster_cost(40, 3) * 30
sp = cluster_cost(40, 3, spot=True) * 30
# Interruptions force reruns; assume 15% of runs redo half their work.
sp_real = sp * (1 + 0.15 * 0.5)
print(f"  on-demand, 30 nights            ${od:8.2f}")
print(f"  spot, 30 nights                 ${sp:8.2f}")
print(f"  spot + 15% partial reruns       ${sp_real:8.2f}"
      f"   saves {1 - sp_real / od:.0%}")
print("  Spot suits retryable batch work, never a job with a hard SLA and no")
print("  checkpointing.")

print("\nONE DASHBOARD QUERY, REFRESHED HOURLY FOR A MONTH")
for label, tb_scanned in [("SELECT * over the whole table", 4.0),
                          ("only 3 of 60 columns", 0.20),
                          ("3 columns + 1 day partition", 0.006)]:
    per_run = tb_scanned * SCAN_PER_TB
    print(f"  {label:30s} {tb_scanned:6.3f} TB"
          f"  ${per_run:6.2f}/run  ${per_run * 24 * 30:8.2f}/mo")
print("  Projection and partition pruning are not micro-optimisations here;")
print("  they are the difference between $22 and $14,400 a month.")

print("\nSTORAGE TIERS FOR 50 TB, AND THE RETRIEVAL TRAP")
GB = 50 * 1024
for share in (0.10, 1.00):
    print(f"  re-reading {share:.0%} of it each month")
    for tier in ("standard", "infrequent", "archive"):
        store = GB * STORAGE_PER_GB_MONTH[tier]
        reread = GB * share * RETRIEVAL_PER_GB[tier]
        print(f"    {tier:11s} hold ${store:8.2f} + read ${reread:8.2f}"
              f" = ${store + reread:8.2f}/mo")
print(f"  plus egress: moving 2 TB to another region costs"
      f" ${2 * 1024 * EGRESS_PER_GB:,.2f}")
print("  Archive is 6x cheaper to hold and the most expensive option once you")
print("  read it monthly. Tier by access pattern, never by age alone.")

  THE SAME WORK, DIFFERENT SHAPES (48 vCPU-hours of compute)
      8 vCPUs x 6.00 h = $  2.04   wall clock   360 min
     48 vCPUs x 1.00 h = $  2.04   wall clock    60 min
    192 vCPUs x 0.25 h = $  2.04   wall clock    15 min
    Per-second billing makes these nearly equal in cost, so buy the
    wall-clock time you need -- as long as the work really scales.

  AMDAHL MEETS THE INVOICE (90% parallel job, 1 vCPU = 10 h)
       1 vCPUs  wall  10.00 h  cost $   0.43  efficiency 100.0%
       4 vCPUs  wall   3.25 h  cost $   0.55  efficiency  76.9%
      16 vCPUs  wall   1.56 h  cost $   1.06  efficiency  40.0%
      64 vCPUs  wall   1.14 h  cost $   3.10  efficiency  13.7%
     256 vCPUs  wall   1.04 h  cost $  11.26  efficiency   3.8%
    Scaling out buys latency, and past a point you pay 20x the money
    for 2x the speed. That can be the right call -- but name it.

  SPOT VS ON-DEMAND FOR A NIGHTLY 40-vCPU, 3-HOUR ETL JOB
    on-demand, 30 nights            $  153.00
    spot, 30 nights                 $   45.90
    spot + 15% partial reruns       $   49.34   saves 68%
    Spot suits retryable batch work, never a job with a hard SLA and no
    checkpointing.

  ONE DASHBOARD QUERY, REFRESHED HOURLY FOR A MONTH
    SELECT * over the whole table   4.000 TB  $ 20.00/run  $14400.00/mo
    only 3 of 60 columns            0.200 TB  $  1.00/run  $  720.00/mo
    3 columns + 1 day partition     0.006 TB  $  0.03/run  $   21.60/mo
    Projection and partition pruning are not micro-optimisations here;
    they are the difference between $22 and $14,400 a month.

  STORAGE TIERS FOR 50 TB, AND THE RETRIEVAL TRAP
    re-reading 10% of it each month
      standard    hold $ 1177.60 + read $    0.00 = $ 1177.60/mo
      infrequent  hold $  640.00 + read $   51.20 = $  691.20/mo
      archive     hold $  204.80 + read $  153.60 = $  358.40/mo
    re-reading 100% of it each month
      standard    hold $ 1177.60 + read $    0.00 = $ 1177.60/mo
      infrequent  hold $  640.00 + read $  512.00 = $ 1152.00/mo
      archive     hold $  204.80 + read $ 1536.00 = $ 1740.80/mo
    plus egress: moving 2 TB to another region costs $184.32
    Archive is 6x cheaper to hold and the most expensive option once you
    read it monthly. Tier by access pattern, never by age alone.

Four lessons, each with a number attached.

Cluster shape is nearly free; parallel efficiency is not. Forty-eight vCPU-hours cost the same whether spread over 6 hours or 15 minutes, so buy latency when you want it. But the second table shows what happens when the job does not scale perfectly: going from 1 to 256 vCPUs on a 90%-parallel job cut wall clock by 9.6x and multiplied cost by 26x. That can be a good trade for a deadline; it is a bad one made by accident.

Spot instances are the largest single lever for batch work, saving 68% net of interruption reruns -- provided the job can be retried. Attach them to checkpointed, rerunnable pipelines and never to a job with a hard SLA.

Query hygiene is a budget line. The same dashboard question cost $21.60 or $14,400 a month depending only on projection and partition pruning. This is the financial version of the previous section on query engines, and it is the argument that gets pruning prioritised.

Storage tiers punish incorrect access assumptions. Archive holds 50 TB for $205 against $1,178 -- and costs $1,741 once you read it all each month. Tier by how the data is actually accessed, which you can measure, rather than by its age, which you cannot infer access from.

Practices that keep costs visible rather than surprising:

• Tag every resource by team, pipeline, and environment. Untagged spend cannot be attributed, and unattributed spend never gets fixed.
• Alert on cost anomalies, not just on failures. A query that silently starts scanning ten times more data is invisible in green dashboards.
• Put a cost estimate in code review for new scheduled queries: bytes scanned times frequency times price.
• Kill idle clusters automatically. A development cluster left running over a weekend is pure loss.
• Reconsider whether the cluster is needed at all. After pruning, many pipelines fit on one machine, and that is usually the largest saving available.

Key Takeaways

• Cloud billing makes performance a financial property of code: compute time, storage, bytes scanned, and data transfer.
• With per-second billing, total vCPU-hours matter far more than cluster shape, so buying latency is cheap when the work scales.
• Imperfect parallelism inverts that: a 90%-parallel job at 256 vCPUs cost 26x more for 9.6x less wall clock.
• Spot instances save roughly 70%, net of reruns, for checkpointed and retryable batch work only.
• On per-TB-scanned pricing, projection and partition pruning changed one dashboard's monthly cost from $14,400 to $21.60.
• Archive tiers are cheap to hold and expensive to read; choose tiers from measured access patterns, not from data age.
• Egress is the commonest surprise on a cloud bill; keep compute in the region where the data lives.
• Tag resources, alert on cost anomalies, estimate cost in review, and auto-terminate idle clusters.

🧪 Practice

1. Estimate the monthly cost of your largest scheduled pipeline from your provider's current prices, broken down by compute, storage, and scan.
2. Compute the break-even interruption rate at which spot instances stop being cheaper than on-demand for a 4-hour job.
3. Take one dashboard query and reduce its bytes scanned by at least 10x through projection and partition pruning, then price both versions.
4. For a 20 TB dataset where 5% is read monthly, work out the cheapest storage tier including retrieval, then repeat for 80% read monthly.
5. Interview: Your team's cloud data bill tripled in a month with no new pipelines. How do you find the cause? (Hint: the same query can cost much more than it did before if the data behind it grew or its partition filter stopped matching.)

---

<a id="15-productionizing-models-and-mlops"></a>