14. Large-Scale Data and Intelligent Systems

Every system that survives long enough accumulates two things it was not originally designed for: an enormous amount of historical data, and a demand to make decisions from it. This chapter covers the platforms that store analytical data, the batch and streaming engines that process it, the search and recommendation systems that turn it into ranked results, and the ML serving infrastructure that puts models on the request path. The emphasis throughout is on the systems around the data and the models — pipelines, freshness, correctness, and rollout — rather than on the statistics inside them.

<a id="141-data-platform-architecture"></a>

14.1 Data Platform Architecture

Analytical systems fail differently from transactional ones: not with a dropped write, but with a number that is quietly wrong, or a query that costs $400. This subchapter covers why analytical workloads need their own storage architecture, the warehouse and lakehouse designs that serve them, how transformation pipelines are ordered, how analytical schemas are modelled, and the quality and lineage controls that make the output trustworthy.

OLTP vs. OLAP Workloads

A single database engine cannot be excellent at both of the jobs a company needs from its data. The first job is running the product: "insert this order", "fetch user 42's cart", "decrement this inventory count". These are OLTP (online transaction processing) workloads — enormous numbers of tiny operations, each touching a handful of rows, each needing to be correct and fast right now. The second job is understanding the product: "what was revenue by country by week for the last two years, split by acquisition channel?" That is OLAP (online analytical processing) — a small number of enormous queries, each touching billions of rows but only a few columns.

The reason these need different systems is physical, not political. Storage reads in blocks. A row-oriented store keeps all of a row's columns adjacent, so fetching one whole order costs one block read — perfect for OLTP, terrible for scanning one column across a billion rows, because every block you read is mostly columns you did not want. A column-oriented store keeps each column contiguous, so scanning revenue reads only revenue blocks. It also compresses far better, because a column holds values of one type with heavy repetition: a country column across a billion rows might hold 200 distinct values, which dictionary encoding turns into roughly one byte each.

 ROW STORE (OLTP)                     COLUMN STORE (OLAP)

 block 1: [id|user|country|cents]     block 1: [id, id, id, id, ...]
 block 2: [id|user|country|cents]     block 2: [user, user, user, ...]
 block 3: [id|user|country|cents]     block 3: [country, country, ...]  <-- only
 block 4: [id|user|country|cents]     block 4: [cents, cents, cents, ...] <-- these

 "get order 903"      -> 1 block      "get order 903"      -> 4 scattered reads
 "SUM(cents) by       -> reads ALL    "SUM(cents) by       -> reads 2 of 4
  country"               4 blocks      country"               column files

The architectural consequence is that you do not point dashboards at the production database. Analytical queries scan enough data to evict the OLTP working set from cache, and hold snapshots long enough to bloat the write path — a report can take down checkout. Analytical data is instead replicated into a separate system tuned for scans, with a deliberate freshness lag.

Dimension	OLTP	OLAP
Unit of work	One row / small transaction	Billions of rows, few columns
Query rate	Thousands per second	Thousands per day
Latency target	Single-digit ms	Seconds to minutes
Storage layout	Row-oriented, B-tree indexes	Columnar, sorted, compressed
Concurrency	Very high, short-lived	Low, long-lived
Schema	Normalized (3NF)	Denormalized (star schema)
Correctness need	ACID per transaction	Consistent snapshot per query
Typical systems	PostgreSQL, MySQL, DynamoDB	BigQuery, Snowflake, ClickHouse

# Why columnar wins: a rough cost model for one analytical query.
ROWS          = 2_000_000_000
BYTES_PER_ROW = 400          # full row width across ~30 columns
QUERY_COLS    = 3            # the query references only 3 of them
COL_BYTES     = 12           # combined width of those 3 columns, uncompressed
COMPRESSION   = 6            # typical columnar ratio on low-cardinality data
SCAN_GBPS     = 2.0          # effective scan throughput of the engine

row_bytes = ROWS * BYTES_PER_ROW                   # must read every column
col_bytes = ROWS * COL_BYTES / COMPRESSION         # only referenced columns

for label, b in (("row store", row_bytes), ("column store", col_bytes)):
    gb = b / 1e9
    print(f"{label:<13} scans {gb:8.1f} GB -> {gb / SCAN_GBPS:6.1f} s")

# row store     scans    800.0 GB ->  400.0 s
# column store  scans      4.0 GB ->    2.0 s
# A 200x gap that indexing on the row store cannot recover, because the query
# touches most rows — an index improves selectivity, not scan volume.

Modern engines blur the line. HTAP systems (TiDB, SingleStore) and in-database column stores (MySQL HeatWave, columnar extensions for PostgreSQL) maintain both layouts over one dataset. They are a genuine option at moderate scale and remove a pipeline, but they give up the independent scaling and cost profile of a dedicated analytical system, so the split remains the default once analytical volume grows.

Key Takeaways

• OLTP is many tiny row-level operations; OLAP is few huge column-level scans.
• Row stores read whole rows cheaply; column stores read few columns cheaply and compress far better because a column is homogeneous.
• Never run analytics on the production OLTP database: scans evict its cache and long snapshots hurt the write path.
• The separation costs freshness — analytical data is deliberately behind.
• HTAP engines merge both layouts and are viable at moderate scale, but give up independent scaling of the two workloads.

🧪 Practice

1. Take three queries from a product you know and classify each as OLTP or OLAP using rows touched and columns referenced.
2. Estimate bytes scanned for one analytical query under row and columnar layouts, using the cost model above with your own table widths.
3. Interview: A product manager wants a live dashboard reading from the production database. What do you propose instead? (Hint: think about what a full scan does to the OLTP buffer cache, and how stale the dashboard is actually allowed to be.)

Data Warehouses

A data warehouse is the system of record for analysis: one place where data from every source — the product database, payment processors, the CRM, ad platforms, event streams — is cleaned, conformed to shared definitions, and queried with SQL. Its real product is less storage than agreement. When finance, marketing, and engineering each compute "active users" from their own extract, they arrive with three numbers and spend the meeting arguing. A warehouse exists so "active user" is defined once, in one table, that everyone reads.

The defining architectural move of the modern warehouse is the separation of storage and compute. Older MPP appliances bound them together: to store more you bought more machines, and those machines idled between queries. Cloud warehouses keep data in object storage and start compute clusters against it on demand, which changes three things — storage becomes cheap and effectively unbounded, concurrent workloads get isolated clusters over the same data instead of fighting for one, and idle cost approaches zero.

 CLOUD WAREHOUSE: SEPARATED STORAGE AND COMPUTE

        BI tool      data science      scheduled ELT
           |               |                 |
     +-----v-----+   +-----v-----+     +-----v-----+
     | compute   |   | compute   |     | compute   |   <- sized and billed
     | cluster A |   | cluster B |     | cluster C |      independently;
     | (small)   |   | (large)   |     | (medium)  |      autosuspend when idle
     +-----+-----+   +-----+-----+     +-----+-----+
           |               |                 |
           +-------+-------+--------+--------+
                   |                         |
            +------v-------------------------v------+
            |  metadata / catalog (files, stats,     |  <- the "brain": which
            |  partitions, snapshots, permissions)   |     files make a table
            +------+---------------------------------+
                   |
            +------v---------------------------------+
            |  object storage: immutable columnar     |  <- one copy of the data
            |  files (Parquet/ORC), partitioned       |
            +----------------------------------------+

Query performance inside a warehouse comes mostly from not reading data. Three mechanisms do that work: partition pruning (skip whole directories whose partition key falls outside the WHERE clause), file and block skipping (each file carries per-column min/max statistics, so a file whose range cannot match is never opened), and clustering or sort keys (physically ordering rows so those min/max ranges stay tight instead of overlapping).

-- The highest-leverage habit in warehouse SQL: filter the partition column
-- explicitly, and select only the columns you need.

-- BAD: no partition filter, SELECT * -> full-table scan, every column read.
SELECT * FROM events WHERE user_id = 42;

-- GOOD: partition pruning limits files opened; projection limits bytes read.
SELECT event_type, occurred_at                             -- 2 columns, not 30
FROM   events
WHERE  event_date BETWEEN '2026-08-01' AND '2026-08-31'    -- prunes partitions
  AND  user_id = 42;                                       -- min/max skipping

-- Clustering makes the second predicate cheap by keeping user_id ranges tight
-- within each file, so most files are skipped on their statistics alone.
ALTER TABLE events CLUSTER BY (user_id);

Warehouses are usually organized in layers so raw fidelity and business meaning do not fight: a raw/bronze layer holding source data exactly as ingested (never edited, so any downstream bug is replayable), a staging/silver layer with types cast, keys deduplicated, and columns renamed to a standard, and a marts/gold layer of business-facing tables shaped for the questions people actually ask. Cost follows bytes scanned, so the operating discipline is partitioning by time, pre-aggregating anything queried repeatedly, and setting per-query byte limits so one careless SELECT * cannot spend a month's budget (13.4).

Key Takeaways

• A warehouse's main product is agreement on definitions, not storage.
• Separating storage from compute lets workloads scale and be billed independently over a single copy of the data.
• Performance comes from skipping data — partition pruning, min/max file statistics, clustering — not from row-level indexes.
• Layer the warehouse raw -> staging -> marts, and keep raw immutable so any downstream mistake is replayable.
• Bytes scanned is the cost unit: always filter partitions and project columns.

🧪 Practice

1. Rewrite a SELECT * query against a large table so it prunes partitions and projects only needed columns; estimate the bytes saved.
2. Design the three-layer structure for one source system, naming the tables at each layer and the transformation between them.
3. Interview: Why do cloud warehouses separate storage from compute, and what problem does that create that a coupled system did not have? (Hint: think about what now crosses the network on every query, and which caching layer has to compensate.)

Data Lakes and Lakehouses

A warehouse wants structure up front. That is a poor fit for data whose shape is unknown, whose volume is enormous relative to its value per byte, or that is not tabular at all — clickstream JSON, application logs, images, audio, model checkpoints. A data lake answers this by storing files of any format in object storage under an agreed directory layout, applying schema when the data is read rather than when it is written (schema-on-read). Storage is cheap, ingestion is trivial, and nothing is discarded merely because no table exists for it yet.

The failure mode has a name: the data swamp. Without a catalog nobody knows what exists; without schema enforcement a producer changes a field type and downstream jobs break silently; without transactions a reader can observe a half-written set of files; and correcting one record means rewriting a whole partition, which makes deletion requests (11.4) painful. Lakes accumulated data faster than they accumulated trust.

The lakehouse is the response: keep open columnar files in object storage, but put a transaction log over them. Open table formats — Delta Lake, Apache Iceberg, Apache Hudi — maintain an ordered log of metadata commits describing which files constitute the table at each version. That log buys ACID snapshots, schema evolution, row-level updates and deletes, and time travel, without giving up open formats or multi-engine access.

 LAKEHOUSE TABLE = DATA FILES + A COMMIT LOG

 object-store://sales/
   data/  part-0001.parquet  part-0002.parquet  part-0003.parquet  part-0004.parquet
   _log/
     0000.json  -> ADD part-0001, part-0002          (version 1)
     0001.json  -> ADD part-0003                     (version 2)
     0002.json  -> REMOVE part-0002, ADD part-0004   (version 3 = an UPDATE)

 A reader at version 3 sees {0001, 0003, 0004}. A reader that started at version
 2 keeps seeing {0001, 0002, 0003} — a consistent snapshot, unaffected by the
 concurrent rewrite. Nothing is mutated in place; only the log moves forward.

# Row-level deletes over immutable files: the two write modes a table format
# offers, and the trade-off that decides between them.

def copy_on_write(files: dict[str, list[int]], to_delete: set[int]):
    """Rewrite each affected file without the deleted rows.
    Slow writes, fastest reads — readers open only current data files."""
    out = {}
    for name, rows in files.items():
        kept = [r for r in rows if r not in to_delete]
        if len(kept) != len(rows):
            out[name + "-v2"] = kept        # whole file rewritten
        else:
            out[name] = rows                # untouched: no cost
    return out

def merge_on_read(files: dict[str, list[int]], to_delete: set[int]):
    """Write a small delete marker instead.
    Fast writes, slower reads — every reader subtracts markers until
    compaction folds them into the data files."""
    return files, {"delete_markers": sorted(to_delete)}

files = {"part-0001": [1, 2, 3], "part-0002": [4, 5, 6]}
print(copy_on_write(files, {5}))     # part-0002 rewritten as part-0002-v2
print(merge_on_read(files, {5}))     # data untouched; marker file added
# Copy-on-write for read-heavy tables with occasional corrections;
# merge-on-read for streaming upserts where write latency dominates.

Property	Data lake (raw files)	Warehouse	Lakehouse
Storage format	Any	Proprietary	Open columnar
Schema	On read	On write	On write, evolvable
Transactions	None	Yes	Yes (log-based)
Row-level updates	Rewrite partition	Native	Native
Engine lock-in	None	High	Low
Time travel	Manual snapshots	Limited retention	Native, by version
Cost per TB	Lowest	Highest	Low
Governance maturity	Weakest	Strongest	Improving

For most organizations the answer is not one of the three but a boundary: raw and semi-structured data lands in lakehouse tables, curated high-concurrency business marts may live in a warehouse, and one catalog describes both so a query planner and a permission system can see the whole estate.

Key Takeaways

• A lake stores anything cheaply with schema-on-read; without a catalog and contracts it degrades into a swamp nobody trusts.
• A lakehouse adds a transaction log over open columnar files, giving ACID snapshots, schema evolution, row-level writes, and time travel.
• Copy-on-write favours read speed; merge-on-read favours write speed and depends on compaction to keep reads fast.
• Open formats keep multiple engines usable over one copy of the data.
• Time travel is what makes "reproduce yesterday's number" possible.

🧪 Practice

1. Sketch the directory and log layout for a table receiving daily appends plus weekly corrections, and state which write mode you would choose.
2. List three governance controls that stop a lake becoming a swamp, and name an owner for each.
3. Interview: A GDPR request requires deleting one user's rows from a 50 TB lake table. How do you do it efficiently? (Hint: think about what makes those rows findable without a full scan, and which write mode avoids rewriting every file that user ever appeared in.)

ETL vs. ELT Pipelines

Data rarely arrives in the shape analysis needs, so every platform runs three steps: extract from sources, transform into conformed types and business definitions, and load into the analytical store. The interesting question is the order of the last two, and that order follows directly from what compute costs.

ETL transforms before loading, in a dedicated processing tier. It was the default when warehouse compute was scarce and bought by the year: you did the heavy lifting on cheaper machines and loaded only the finished result. The cost is that the warehouse never sees raw data, so a transformation bug requires re-extracting from the source — and the source may no longer hold it.

ELT loads raw data first and transforms inside the warehouse with SQL. It became the default once cloud compute turned elastic and per-second billed, and its advantages compound: the raw layer is preserved so transformations can be replayed and fixed without touching sources, transformations are SQL that analysts can read and review, and one engine runs both the pipeline and the queries, so there is a single system to scale and monitor.

 ETL                                    ELT
 source                                 source
   | extract                              | extract
   v                                      v
 [ processing tier ]                    [ warehouse: RAW layer ]  <- immutable
   | transform (Spark/Python)             | transform (SQL, in-warehouse)
   v                                      v
 [ warehouse ]  <- final data only      [ STAGING ] -> [ MARTS ]

 Bug found? Re-extract from source.     Bug found? Re-run SQL over raw.
 The source may have rotated the data.  Minutes, no source involvement.

ETL still wins in specific cases: when data must be masked or tokenized before it may legally enter the warehouse (11.4), when the transformation is not expressible in SQL (image decoding, model inference), or when volume must be reduced before crossing an expensive network boundary.

Orthogonal to the ordering is the load pattern. A full refresh recomputes the whole table — simple, and fine until the table is large. Incremental loading processes only new or changed rows, and needs a watermark column plus idempotent writes, because pipelines get retried.

from datetime import datetime, timedelta, timezone

def incremental_window(last_watermark: datetime, now: datetime,
                       lateness: timedelta = timedelta(hours=2)):
    """Re-read a safety overlap so late-arriving records are not skipped.
    The overlap is only safe because the write below is idempotent."""
    return last_watermark - lateness, now

lo, hi = incremental_window(
    datetime(2026, 8, 30, 12, tzinfo=timezone.utc),
    datetime(2026, 8, 31, 12, tzinfo=timezone.utc))

print(f"""
MERGE INTO marts.orders AS t
USING (
  SELECT * FROM raw.orders
  WHERE updated_at >= '{lo:%Y-%m-%d %H:%M}'   -- overlap absorbs late arrivals
    AND updated_at <  '{hi:%Y-%m-%d %H:%M}'   -- exclusive bound: no gaps, no
) AS s                                        -- double-counting at the seam
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET *                -- re-running changes nothing
WHEN NOT MATCHED THEN INSERT *;               -- => the job is safe to retry
""")

Whichever ordering you choose, pipelines must be idempotent and replayable. Idempotent means running the same job twice leaves the same table — achieved with MERGE on a natural key, or by deleting and rewriting a whole partition rather than appending. Replayable means any output can be rebuilt from inputs you still hold, which is precisely why the raw layer is never edited.

Aspect	ETL	ELT
Transform location	Separate processing tier	Inside the warehouse
Raw data retained	Usually not	Yes, by design
Language	Python/Scala/Spark	SQL (plus dbt-style tooling)
Fixing a logic bug	Re-extract from source	Re-run SQL over raw
Best when	Masking, non-SQL work	Elastic warehouse compute
Governance	Sensitive data never lands	Raw data must be protected

Key Takeaways

• ELT loads raw first and transforms in-warehouse; ETL transforms first and is chosen for masking, volume reduction, or non-SQL work.
• An immutable raw layer makes transformation bugs replayable without touching source systems.
• Incremental loads need a watermark plus a lateness overlap, and that overlap is only safe if writes are idempotent.
• MERGE on a natural key, or whole-partition rewrite, makes retries harmless.
• ELT moves sensitive data into the warehouse, raising the governance bar.

🧪 Practice

1. Convert a full-refresh job into an incremental one: choose the watermark column, the overlap, and the idempotent write strategy.
2. Identify a pipeline you know that is not idempotent, and describe exactly what a mid-run retry would corrupt.
3. Interview: A nightly pipeline failed halfway and was restarted; some rows are now duplicated. What design would have prevented it? (Hint: think about what makes a write's result depend only on the input, not on how many times it ran.)

Data Modeling for Analytics

Transactional schemas are normalized to avoid update anomalies: each fact lives in exactly one place, so a change touches one row. That is right when writes dominate. Analytical schemas optimize for the opposite verb, and their enemy is not redundancy but the join. "Revenue by product category by region by month" against a fully normalized schema may require eight joins, each of which the engine must plan, shuffle, and materialize.

The dominant analytical pattern is the star schema: one central fact table holding measurements at a declared grain (one row per order line, one row per page view) surrounded by dimension tables holding descriptive attributes (product, customer, date, store). Facts are numeric, additive, and enormous; dimensions are textual, small, and used for filtering and grouping. Every query becomes one fact table plus a few single-hop dimension joins — a shape engines optimize aggressively with broadcast joins, since dimensions fit in memory.

 STAR SCHEMA

        dim_date                     dim_customer
        +------------+               +----------------+
        | date_key PK|               | customer_key PK|
        | day, week  |               | name, segment  |
        | month, qtr |               | country, tier  |
        +-----+------+               +--------+-------+
              |                               |
              |     fact_order_line           |
              |     +---------------------+   |
              +---->| date_key        FK  |<--+
                    | customer_key    FK  |
              +---->| product_key     FK  |
              |     | store_key       FK  |<--+
              |     |---------------------|   |
              |     | quantity     MEASURE|   |   GRAIN: one row per order line.
              |     | net_amount   MEASURE|   |   Write it down first; every
              |     | discount     MEASURE|   |   measure must be true at it.
              |     +---------------------+   |
        dim_product                     dim_store

Declaring the grain first is the single most important modelling decision. If one row is one order line, net_amount is additive but order_shipping_fee is not — summing it across lines multiplies the fee by the number of lines. Hence the standard classification of measures: additive (summable across every dimension), semi-additive (summable across some, e.g. an account balance across accounts but not across time), and non-additive (ratios and percentages, which must be recomputed from their components rather than averaged).

Dimensions change over time, and how you handle that determines whether history stays honest. A customer moves from "SMB" to "Enterprise": do last year's orders now report as Enterprise revenue? Slowly changing dimension types answer this — Type 1 overwrites (history is rewritten), Type 2 inserts a new version row with validity dates and keeps the old one (history preserved), Type 3 keeps a "previous value" column.

-- Type 2 SCD: one row per version, with a surrogate key so each fact points at
-- the version that was current when the fact occurred.
CREATE TABLE dim_customer (
  customer_key   BIGINT,      -- surrogate PK: unique per VERSION
  customer_id    STRING,      -- natural key: stable across versions
  segment        STRING,
  valid_from     TIMESTAMP,
  valid_to       TIMESTAMP,   -- NULL (or 9999-12-31) for the current row
  is_current     BOOLEAN
);

-- 5001  C-42  SMB         2024-01-01 -> 2026-03-15   is_current = false
-- 7310  C-42  Enterprise  2026-03-15 -> NULL         is_current = true
--
-- A 2025 order stores customer_key 5001, so it still reports as SMB revenue.
-- Joining on customer_id instead would silently restate all of history.

SELECT d.segment, SUM(f.net_amount) AS revenue
FROM   fact_order_line f
JOIN   dim_customer  d ON f.customer_key = d.customer_key   -- version-accurate
JOIN   dim_date      t ON f.date_key     = t.date_key
WHERE  t.year = 2025
GROUP  BY d.segment;

A snowflake schema normalizes dimensions further (product -> category -> department as separate tables), saving storage at the cost of extra joins; it is rarely worth it now that storage is cheap. Wide denormalized tables go the other way — one flat table with dimensions pre-joined — and are common in columnar engines where repeated strings compress to almost nothing; they trade storage and update cost for zero-join queries. Choose by how often dimensions change and how many tools consume the table.

Key Takeaways

• Analytical modelling optimizes reads; the cost to minimize is joins, not redundancy.
• A star schema is one large fact table at a declared grain plus small dimensions joined one hop away.
• Declare the grain before anything else; every measure must be valid at it.
• Classify measures as additive, semi-additive, or non-additive — averaging a ratio is a classic silent error.
• Type 2 dimensions preserve history; joining on the natural key instead of the version key silently restates the past.

🧪 Practice

1. Model a star schema for an e-commerce checkout: declare the grain, list the fact measures, and name four dimensions.
2. Take a metric you know and classify each input as additive, semi-additive, or non-additive; show which one breaks under summing.
3. Interview: A customer's country changes. Should past orders report under the old or the new country? (Hint: there is no universal answer — ask who consumes the report and whether they need history as it was, or as it is.)

Data Quality and Lineage

A dashboard that is down is a nuisance; a dashboard that is confidently wrong is a liability, because decisions get made from it. Analytical systems fail silently by default: a source adds a column, a currency changes units, a job skips a partition — and the query still returns a number. Data quality engineering is the practice of making those failures loud.

The mechanism is assertions that run inside the pipeline, not reports generated after it. Checks fall into recognisable families: freshness (was this table updated within its expected interval?), volume (is today's row count within a plausible band of recent days?), schema (have columns or types changed?), uniqueness and referential integrity (are keys unique, do foreign keys resolve?), and distribution (has the null rate or the mean moved beyond a threshold?).

 QUALITY GATE INSIDE THE PIPELINE

  extract --> load raw --> transform --> [ CHECKS ] --+-- pass --> publish
                                              |       |            (swap the
                                              |       +-- fail -->  pointer to
                                              |                     the new
                                     block publication,             snapshot)
                                     page the owner,
                                     leave consumers on
                                     yesterday's good data

  The key property: consumers never see a bad table, because publication is a
  metadata swap that happens only after checks pass. For nearly every analytical
  consumer, stale-but-correct beats fresh-but-wrong.

from dataclasses import dataclass
from typing import Callable

@dataclass
class Check:
    name: str
    sql: str
    severity: str          # "error" blocks publication; "warn" only alerts

CHECKS = [
    Check("freshness", """
        SELECT DATEDIFF('hour', MAX(updated_at), CURRENT_TIMESTAMP()) > 26
        FROM staging.orders""", "error"),            # daily job + 2h grace
    Check("row_volume", """
        SELECT ABS(COUNT(*) - (SELECT AVG(cnt) FROM daily_counts))
               > 4 * (SELECT STDDEV(cnt) FROM daily_counts)
        FROM staging.orders WHERE load_date = CURRENT_DATE()""", "error"),
    Check("unique_key", """
        SELECT COUNT(*) > 0 FROM (
          SELECT order_id FROM staging.orders
          GROUP BY order_id HAVING COUNT(*) > 1)""", "error"),
    Check("fk_resolves", """
        SELECT COUNT(*) > 0 FROM staging.orders o
        LEFT JOIN dim_customer c ON o.customer_key = c.customer_key
        WHERE c.customer_key IS NULL""", "error"),
    Check("null_rate", """
        SELECT AVG(CASE WHEN country IS NULL THEN 1 ELSE 0 END) > 0.02
        FROM staging.orders""", "warn"),             # drift, not corruption
]

def run_gate(checks: list[Check], execute: Callable[[str], bool]) -> bool:
    blocking: list[str] = []
    for c in checks:
        if execute(c.sql):                    # the SQL returns True on FAILURE
            print(f"[{c.severity.upper()}] {c.name}")
            if c.severity == "error":
                blocking.append(c.name)
    if blocking:
        print(f"publication blocked by: {', '.join(blocking)}")
    return not blocking

Lineage is the other half: a graph recording which datasets, columns, and jobs produced each output. Its value appears at exactly two moments. When a number looks wrong, lineage answers "what feeds this?" and turns a day of archaeology into a two-minute traversal. When a source is about to change, it answers the reverse — "what breaks if I drop this column?" — which is the difference between a planned migration and an outage. Column-level lineage captured by parsing the SQL, rather than maintained by hand, is what makes impact analysis precise instead of alarmist.

Around both sit data contracts: explicit, versioned agreements between a producing service and its consumers covering schema, semantics, freshness, and the process for breaking changes. Contracts move the fix upstream — the producer is prevented from shipping an incompatible change — instead of leaving every downstream team to detect breakage independently. Pair them with ownership: every important table needs a named owner, an SLA, and an on-call path (12.2), or the checks simply page a channel nobody reads.

Key Takeaways

• Analytical systems fail silently; quality checks exist to make wrong data loud rather than plausible.
• Run checks as a publication gate inside the pipeline — stale-but-correct beats fresh-but-wrong.
• Cover freshness, volume, schema, uniqueness, referential integrity, and distribution; separate blocking errors from warnings.
• Lineage answers both "what feeds this number?" and "what breaks if I change this?" — derive it from SQL parsing, not from documentation.
• Data contracts push the fix upstream to the producer; ownership and SLAs make the alerts actionable.

🧪 Practice

1. Write freshness, volume, and uniqueness checks for a table you know, including thresholds and the reasoning behind them.
2. Draw the lineage graph for one dashboard metric back to its sources, and mark which upstream change would break it.
3. Interview: A key metric dropped 30% overnight. How do you decide whether it is a real change or a data bug? (Hint: think about what else would have moved if it were real, and which pipeline signals distinguish "fewer events" from "fewer events ingested".)

<a id="142-batch-and-stream-processing"></a>

14.2 Batch and Stream Processing

Storing data is the easy half; computing over it at scale is where the distributed-systems problems return. This subchapter covers the execution model that made large-scale batch processing tractable, the query engines that generalized it, the delivery and state semantics that make streaming results trustworthy, the windowing and watermark machinery that gives unbounded data a notion of "done", and the two architectures that combine batch and streaming.

MapReduce Model

Processing a petabyte on one machine is impossible, and hand-writing a distributed program for every analysis is impractical — you would rewrite partitioning, scheduling, retries, and machine failure handling each time. MapReduce's insight was that a large fraction of data processing fits one skeleton, so the framework can own all the hard distributed parts and let the user supply two ordinary functions.

The skeleton has three phases. Map runs a function over each input record independently, emitting key-value pairs — this is embarrassingly parallel, one task per input split, scheduled near the data so bytes do not cross the network. Shuffle groups every emitted pair by key, moving records across the network so that all values for a key land on one machine. Reduce runs a function over each key's grouped values, emitting the final output.

The analogy is counting votes nationally. Each polling station tallies its own ballots (map), tallies are routed by candidate to that candidate's national desk (shuffle), and each desk sums its pile (reduce). No station needs to see the whole country, and if one station's count is lost you re-run only that station.

 MAPREDUCE: WORD COUNT

 input splits        MAP (parallel)          SHUFFLE            REDUCE
 +-----------+   +-------------------+   by key, over    +------------------+
 | "a b a"   |-->| (a,1)(b,1)(a,1)   |-\  the network    | a: [1,1,1] -> 3  |
 +-----------+   +-------------------+  \--------------->+------------------+
 | "b c"     |-->| (b,1)(c,1)        |-----------------> | b: [1,1,1] -> 3  |
 +-----------+   +-------------------+                   +------------------+
 | "a b"     |-->| (a,1)(b,1)        |-----------------> | c: [1] -> 1      |
 +-----------+   +-------------------+                   +------------------+

 The SHUFFLE is the expensive phase: it is all-to-all network traffic plus a
 disk sort. Every serious optimisation is about shuffling less.

from collections import defaultdict
from typing import Iterator

def map_fn(line: str) -> Iterator[tuple[str, int]]:
    for word in line.split():
        yield word, 1

def combine_fn(key: str, values: list[int]) -> tuple[str, int]:
    """A COMBINER is a reducer run locally on the mapper's output before the
    shuffle. Valid only when the reduce is commutative and associative (sum,
    max, count), and it can cut shuffle bytes by orders of magnitude."""
    return key, sum(values)

def reduce_fn(key: str, values: list[int]) -> tuple[str, int]:
    return key, sum(values)

def run(splits: list[list[str]], use_combiner: bool = True):
    shuffled: dict[str, list[int]] = defaultdict(list)
    shuffle_records = 0
    for split in splits:                     # each split = one map task
        local: dict[str, list[int]] = defaultdict(list)
        for line in split:
            for k, v in map_fn(line):
                local[k].append(v)
        emitted = ([combine_fn(k, vs) for k, vs in local.items()]
                   if use_combiner
                   else [(k, v) for k, vs in local.items() for v in vs])
        shuffle_records += len(emitted)
        for k, v in emitted:
            shuffled[k].append(v)            # this crosses the network
    result = dict(reduce_fn(k, vs) for k, vs in shuffled.items())
    return result, shuffle_records

data = [["a b a", "b c"], ["a b", "c c c"]]
print(run(data, use_combiner=True))    # ({'a': 3, 'b': 3, 'c': 4}, 6 records)
print(run(data, use_combiner=False))   # ({'a': 3, 'b': 3, 'c': 4}, 10 records)

Two properties made the model durable. Fault tolerance is cheap because map and reduce tasks are deterministic and side-effect-free: a failed task is simply re-run on another machine, and at cluster scale machines fail constantly. Scalability is horizontal: more input means more map tasks, and the framework schedules them without user involvement.

Its limitations drove everything that followed. Every stage materializes to disk, so multi-stage jobs — and especially the iterative algorithms common in machine learning — pay repeated write-read cycles. The API is also restrictive: real analysis is a DAG of operations, not one map and one reduce, so users chained jobs by hand. Spark and similar engines generalize the model to a DAG with in-memory intermediate data and lineage-based recovery (recompute a lost partition from its parents rather than replicating it), and Flink extends the DAG to unbounded input. The mental model still matters: map is parallel and cheap, shuffle is the expensive part, and reduce is where skew hurts — one key holding 40% of the data ("hot key" skew) means one reducer runs long after every other has finished, which is the most common cause of a job that is 95% done for an hour.

Key Takeaways

• MapReduce lets the framework own partitioning, scheduling, and failure recovery while users supply two pure functions.
• Map is parallel and local; shuffle is all-to-all network plus sort; reduce aggregates per key.
• Combiners cut shuffle volume dramatically, but only for commutative and associative reductions.
• Fault tolerance comes from deterministic, side-effect-free tasks that can be re-run anywhere.
• Key skew is the classic performance failure: one hot key serializes the job.
• Modern engines keep the model but generalize to in-memory DAGs.

🧪 Practice

1. Express "count distinct users per country per day" as map, combine, and reduce functions.
2. Explain why computing a median is harder in this model than computing a sum, and give a workable approach.
3. Interview: A job is stuck at 99% for an hour with one task running. What is happening and how would you fix it? (Hint: think about how records are assigned to reducers, and what you can do to a key whose share of the data is enormous.)

Distributed Query Engines

Writing map and reduce functions by hand is the assembly language of data processing. Distributed query engines — Spark SQL, Trino/Presto, Flink SQL, BigQuery, ClickHouse — let you write declarative SQL over data spread across hundreds of machines and take responsibility for turning it into an efficient distributed plan. The value is that the engine can rewrite your query, whereas it cannot rewrite your Python.

Execution proceeds through a pipeline. The parser produces a logical plan; the optimizer rewrites it using both rules (push filters toward the scan, prune unused columns, collapse projections) and cost-based decisions informed by statistics (which table is small enough to broadcast, which join order minimizes intermediate rows); the scheduler splits the physical plan into stages separated by shuffles and dispatches tasks to workers.

 QUERY -> PLAN -> STAGES

 SELECT c.country, SUM(o.amount)
 FROM orders o JOIN customers c USING (customer_id)
 WHERE o.order_date >= '2026-08-01'
 GROUP BY c.country;

      LOGICAL                     OPTIMIZED                  PHYSICAL STAGES
   +-----------+               +-----------+              stage 1: scan orders
   | aggregate |               | aggregate |                (filter + project
   +-----+-----+               +-----+-----+                 pushed into scan)
         |                           |                     stage 2: scan customers
   +-----+-----+               +-----+-----+                 (small -> broadcast)
   |   join    |               |   join    |               stage 3: hash join +
   +--+-----+--+               +--+-----+--+                 partial aggregate
      |     |                     |     |                  --- shuffle by country
   +--+-+ +-+---+            +----+-+ +-+-----+            stage 4: final aggregate
   |scan| |scan |            |scan  | |scan   |
   |ord | |cust |            |+filt | |+proj  |   <- FILTER PUSHDOWN: the single
   +----+ +-----+            |+proj | +-------+      biggest win, because it
   (filter applied late)     +------+                shrinks data before shuffle

Join strategy is where an engine earns its keep. A broadcast (map-side) join ships the small side to every worker and needs no shuffle of the large side — the fastest option when one side fits in memory. A shuffle hash join repartitions both sides by the join key so matching rows meet on the same worker. A sort-merge join sorts both sides and streams them together, which handles data larger than memory. Choosing wrongly — broadcasting a table that is not small, or shuffling one that is — is the most common cause of a query that runs for an hour instead of a minute.

def choose_join(left_rows: int, left_bytes: int,
                right_rows: int, right_bytes: int,
                broadcast_limit_bytes: int = 10_000_000,
                workers: int = 100) -> str:
    small, small_bytes = ((("right"), right_bytes) if right_bytes <= left_bytes
                          else (("left"), left_bytes))
    if small_bytes <= broadcast_limit_bytes:
        # Cost: replicate small side to every worker; large side never moves.
        cost = small_bytes * workers
        return f"broadcast {small} side (~{cost/1e6:.0f} MB network total)"
    # Otherwise both sides must be repartitioned by the join key.
    cost = left_bytes + right_bytes
    return f"shuffle hash join (~{cost/1e9:.1f} GB shuffled)"

print(choose_join(2_000_000_000, 800_000_000_000, 50_000, 4_000_000))
# broadcast right side (~400 MB network total)
print(choose_join(2_000_000_000, 800_000_000_000, 900_000_000, 300_000_000_000))
# shuffle hash join (~1100.0 GB shuffled)

Two operational realities matter more than plan theory. First, statistics decay: a cost-based optimizer with stale table statistics makes confident bad choices, which is why modern engines add adaptive execution — re-planning mid-query once actual partition sizes are known. Second, skew defeats the optimizer's uniformity assumption; the standard remedy is salting, appending a random suffix to the hot key so its rows spread across many partitions, then aggregating twice.

-- Salting a skewed join key: split the hot key across 32 partitions.
WITH salted_facts AS (
  SELECT *, CONCAT(customer_id, '#',
                   CAST(FLOOR(RAND() * 32) AS STRING)) AS salted_key
  FROM fact_events                          -- one customer holds 40% of rows
),
salted_dim AS (
  SELECT d.*, CONCAT(d.customer_id, '#', CAST(n AS STRING)) AS salted_key
  FROM dim_customer d
  CROSS JOIN UNNEST(GENERATE_ARRAY(0, 31)) AS n   -- replicate the small side
)
SELECT f.event_type, COUNT(*)                     -- join now spreads evenly
FROM salted_facts f JOIN salted_dim d USING (salted_key)
GROUP BY f.event_type;

Engines also differ in intent. Trino and similar MPP engines are tuned for interactive queries across federated sources; Spark is tuned for large batch pipelines with heavy transformation; ClickHouse and Druid are tuned for low-latency aggregation over pre-shaped data. Picking the engine whose design point matches the workload matters more than tuning the wrong one.

Key Takeaways

• Declarative SQL lets the engine rewrite the plan; imperative code prevents it.
• Filter and projection pushdown before any shuffle is the largest single win.
• Broadcast joins avoid shuffling the large side but only work when one side genuinely fits in memory; otherwise shuffle hash or sort-merge.
• Cost-based optimizers depend on fresh statistics; adaptive execution re-plans using observed sizes.
• Skew breaks uniformity assumptions — salt hot keys and aggregate twice.
• Match the engine to the workload: interactive, batch, or low-latency aggregation.

🧪 Practice

1. Take a slow query you know, read its plan, and identify the stage that moves the most data.
2. Given a 900 GB fact table and a 6 MB dimension, state the join strategy and the network cost under each alternative.
3. Interview: A join is slow because one key dominates. Describe two fixes. (Hint: one changes the key distribution, the other avoids the shuffle entirely by making the other side small enough to copy.)

Stream Processing Semantics

Batch processing has a comfortable property: the input is finite, so a job can finish and be re-run to correct a mistake. Streams are unbounded, so there is no "finished" and no clean re-run — the system must produce continuously correct results while machines fail underneath it. That forces you to be explicit about two questions people usually conflate: how many times can a record be delivered, and how many times can it affect the result.

Delivery guarantees are the transport-level answer. At-most-once acknowledges before processing: fast, and records vanish on crash. At-least-once acknowledges after processing: nothing is lost, but a crash between processing and acknowledgement causes reprocessing, so duplicates are guaranteed to occur eventually. Exactly-once is impossible as a delivery guarantee across an unreliable network — the acknowledgement itself can be lost — so what real systems provide is exactly-once processing semantics: records may be delivered many times, but their effect on state and output is applied once.

 WHERE DUPLICATES COME FROM

  source ---(1) read---> [ processor: update state ] ---(2) write---> sink
     ^                            |                                     |
     |                     (3) commit offset                            |
     +---- crash between (2) and (3): the record is re-read and its ----+
           effect is applied a SECOND time, unless (2) and (3) commit
           atomically or the sink write is idempotent.

Exactly-once is achieved in one of two ways. The first is a transactional commit: state updates, output writes, and the source offset advance atomically, so a crash rolls back to a consistent point — Flink's checkpoint barriers with a two-phase sink commit, or Kafka Streams' transactional producer, work this way. The second, usually cheaper, is an idempotent sink: write with a deterministic key so applying the same record twice leaves the same result. If the sink is idempotent, at-least-once delivery already gives you effectively-once results, and this is the pragmatic choice in most production systems.

class ExactlyOnceProcessor:
    """Effectively-once via an idempotent sink plus a dedupe window.
    The dedupe set bounds duplicates from retries; the sink upsert makes even
    a missed dedupe harmless."""

    def __init__(self, sink, window_size: int = 100_000):
        self.sink = sink
        self.seen: dict[str, int] = {}      # event_id -> offset, bounded
        self.window_size = window_size

    def process(self, record) -> bool:
        if record.event_id in self.seen:
            return False                     # duplicate from a retry: drop
        result = transform(record)
        # Deterministic key => replaying this write changes nothing.
        self.sink.upsert(key=record.event_id, value=result)
        self.seen[record.event_id] = record.offset
        if len(self.seen) > self.window_size:
            # Evict the oldest half; duplicates older than the window are
            # still absorbed by the upsert, just less cheaply.
            cutoff = sorted(self.seen.values())[len(self.seen) // 2]
            self.seen = {k: v for k, v in self.seen.items() if v >= cutoff}
        return True

def transform(record):
    return {"user": record.user_id, "value": record.value * 2}

The other pillar is state. A stateless map is easy to recover — just re-read. A stateful operator (running counts, session tracking, joins across streams) holds data that must survive failure, so streaming engines keep operator state in an embedded store (often RocksDB) and periodically checkpoint it to durable storage. Recovery restores the last checkpoint and rewinds the source to the offset recorded in it. This is why the source must be replayable: a log like Kafka or Kinesis works, a plain HTTP push does not.

Guarantee	Records lost?	Duplicates?	Cost
At-most-once	Possible	No	Lowest latency, no bookkeeping
At-least-once	No	Yes	Cheap; needs idempotent sinks
Exactly-once	No	No effect	Transactions + checkpoints;
			higher latency and complexity

Key Takeaways

• Separate delivery guarantees from processing effects: exactly-once delivery is impossible, exactly-once effect is not.
• At-least-once plus an idempotent sink is the practical default and is far cheaper than distributed transactions.
• True exactly-once needs state, output, and offset to commit atomically.
• Stateful operators need periodic checkpoints plus a replayable source; without replay there is no recovery.
• Checkpoint interval trades recovery time against steady-state overhead.

🧪 Practice

1. For a stream job you know, identify the sink and state whether it is idempotent; if not, design a key that would make it so.
2. Trace what happens on a crash between writing output and committing the offset, under each of the three guarantees.
3. Interview: Your pipeline claims exactly-once but the downstream table shows duplicates. Where would you look? (Hint: the guarantee only spans the operators inside the engine — think about what happens at the boundary where data leaves it.)

Windowing Strategies

Aggregation needs a finite set of rows, but a stream never ends — "the average order value" over an infinite stream is not a computable quantity. Windowing solves this by slicing the unbounded stream into bounded chunks that can be closed and emitted. Choosing the window shape is a modelling decision about what question you are asking, not a tuning knob.

Tumbling windows are fixed-size and non-overlapping: every event belongs to exactly one window. They are the right choice for periodic reporting ("orders per minute") because the results partition the timeline cleanly and sum correctly.

Sliding (hopping) windows are fixed-size but advance by a smaller step, so windows overlap and each event lands in several. They answer "at any moment, what was the total over the last 5 minutes?" — the natural shape for alerting, because a tumbling window can split a burst across a boundary and miss the threshold. Cost scales with the overlap factor: a 5-minute window advancing every 30 seconds puts every event in 10 windows.

Session windows have no fixed size. They group events separated by less than a gap of inactivity, so the window length is defined by the data. This is the model for user behaviour: a session ends when someone stops interacting, not when a clock ticks over.

 events:   a   b        c d     e            f g h        i
 time:  ---1---3--------8-9----13-----------22-23-24------31--->

 TUMBLING (size 10)      [0-------10)[10------20)[20------30)[30---
   contents               a b c d      e            f g h      i

 SLIDING (size 10, hop 5) [0------10)
                              [5-------15)
                                   [10-------20)
                                         [15------25)   <- f,g,h counted in
                                              [20------30)  several windows

 SESSION (gap 4)          [a b]  [c d e]        [f g h]      [i]
   1..3    gap 5 -> close  8..13  gap 9 -> close 22..24       31
   Window length is decided by the data, not by the clock.

from dataclasses import dataclass, field

@dataclass
class Event:
    key: str
    ts: int          # event time, seconds

def tumbling(events: list[Event], size: int) -> dict[int, list[Event]]:
    out: dict[int, list[Event]] = {}
    for e in events:
        start = (e.ts // size) * size            # each event -> exactly 1 window
        out.setdefault(start, []).append(e)
    return out

def sliding(events: list[Event], size: int, hop: int) -> dict[int, list[Event]]:
    out: dict[int, list[Event]] = {}
    for e in events:
        first = ((e.ts - size) // hop + 1) * hop  # earliest window containing e
        for start in range(max(first, 0), e.ts + 1, hop):
            out.setdefault(start, []).append(e)   # duplicated across overlaps
    return out

def sessions(events: list[Event], gap: int) -> list[list[Event]]:
    out: list[list[Event]] = []
    for e in sorted(events, key=lambda x: x.ts):
        if out and e.ts - out[-1][-1].ts <= gap:
            out[-1].append(e)                     # extend the open session
        else:
            out.append([e])                       # inactivity gap: start a new one
    return out

evts = [Event("u", t) for t in (1, 3, 8, 9, 13, 22, 23, 24, 31)]
print(sorted(tumbling(evts, 10)))                 # [0, 10, 20, 30]
print({k: len(v) for k, v in sorted(sliding(evts, 10, 5).items())})
print([[e.ts for e in s] for s in sessions(evts, 4)])
# [[1, 3], [8, 9, 13], [22, 23, 24], [31]]

Two design points follow. Windows must be keyed — you almost always want "per user" or "per device", not a global aggregate — and the number of open windows is keys x windows per key, which is what actually determines memory. Session windows are the most expensive because they can merge retroactively when a late event bridges two open sessions. And every window needs an eviction rule: state for a closed window must be cleared after its allowed lateness expires, or the job leaks memory until it dies days later.

Key Takeaways

• Windowing makes an unbounded stream aggregatable by slicing it into bounded chunks.
• Tumbling windows partition the timeline and are right for periodic reporting.
• Sliding windows overlap and suit alerting, at a cost proportional to size/hop.
• Session windows are data-defined by an inactivity gap and model user behaviour; they may merge when a late event bridges two sessions.
• Open windows are keys x windows per key — that product, plus an eviction rule, determines whether the job stays within memory.

🧪 Practice

1. Choose a window type for each: hourly revenue reporting, "5 failed logins in 2 minutes" alerting, and average time spent per visit.
2. Compute how many windows a single event belongs to for a 10-minute window hopping every 30 seconds, and the memory implication for 1M keys.
3. Interview: Your session-window job's memory grows without bound. What are the likely causes? (Hint: think about keys that never see another event, and about when state for a closed window is actually allowed to be deleted.)

Watermarks and Late Data

Windows raise a question batch systems never face: when is a window done? Events do not arrive in the order they happened. A phone goes through a tunnel and uploads twenty minutes of events at once; a partition is re-consumed after a restart; one region's network is slow. So the system must distinguish event time (when it happened, stamped at the source) from processing time (when the engine saw it). Aggregating by processing time is easy and produces results that change depending on how fast your cluster was that day — which makes them irreproducible.

A watermark is the engine's assertion that no event with a timestamp earlier than W is expected any more. It is a heuristic, not a fact: it is generated from observed timestamps minus an allowed lateness, and it is the mechanism that lets the system close a window and emit a result. Watermarks make the fundamental trade-off explicit — a conservative watermark waits longer and is more complete; an aggressive one emits sooner and is more likely wrong.

 EVENT TIME vs PROCESSING TIME

 processing time -->
   arrives:   e(10) e(12) e(11)   e(21) e(14)!   e(25)   e(9)!!
                                          ^                ^
                                   late but within    very late, beyond
                                   allowed lateness   lateness -> dropped
                                                      or sent to side output

 watermark = max_seen_event_time - 5

   after e(21): watermark = 16  -> window [0,10) and [10,20) can close
   e(14) arrives after wm=16 -> LATE. If the window is retained, re-emit an
   updated result; otherwise it is dropped.

   +----------------------- allowed lateness (window kept in state) ----+
   window closes  -----------------------------------------------> state evicted

class WatermarkTracker:
    """Bounded-out-of-orderness watermark: the standard heuristic."""

    def __init__(self, max_out_of_order_s: int, allowed_lateness_s: int):
        self.max_ooo = max_out_of_order_s
        self.lateness = allowed_lateness_s
        self.max_event_ts = 0
        self.windows: dict[int, int] = {}     # window start -> count
        self.emitted: set[int] = set()

    def watermark(self) -> int:
        return self.max_event_ts - self.max_ooo

    def on_event(self, ts: int, size: int = 10):
        self.max_event_ts = max(self.max_event_ts, ts)
        start = (ts // size) * size
        wm = self.watermark()
        if start + size + self.lateness < wm:
            return "DROPPED (beyond allowed lateness)"   # never silently ignore
        self.windows[start] = self.windows.get(start, 0) + 1
        if start in self.emitted:
            # Window already emitted: this is late data. Emit a correction —
            # only safe if the downstream sink accepts updates (upsert).
            return f"LATE -> re-emit window {start} = {self.windows[start]}"
        return f"buffered in window {start}"

    def advance(self, size: int = 10):
        """Close every window whose end is below the watermark."""
        wm, out = self.watermark(), []
        for start, count in sorted(self.windows.items()):
            if start + size <= wm and start not in self.emitted:
                self.emitted.add(start)
                out.append((start, count))
        return out

t = WatermarkTracker(max_out_of_order_s=5, allowed_lateness_s=20)
for ts in (10, 12, 11, 21):
    t.on_event(ts)
print(t.advance())              # [(10, 3)] -> window [10,20) closes at wm=16
print(t.on_event(14))           # LATE -> re-emit window 10 = 4
print(t.on_event(9))            # window [0,10) is still within lateness

Handling late data is a product decision expressed in three parts. Allowed lateness says how long a window's state is retained for corrections — longer means more memory. Triggers say when to emit: once at the watermark, early and repeatedly for a live-updating dashboard, or again on each late arrival. Accumulation mode says what a re-emission means — a complete restatement of the window (the sink upserts) or a delta to be added (the sink accumulates). Getting these three consistent is what separates a dashboard that converges on the truth from one that double-counts.

One operational trap deserves naming: a watermark advances from observed data, so an idle partition produces no events and holds the global watermark back, stalling every window across the job. Engines provide idleness detection to exclude such partitions; without it, one quiet shard silently freezes all output.

Key Takeaways

• Event time is when it happened, processing time is when you saw it; only event time gives reproducible results.
• A watermark is a heuristic assertion that no earlier events remain — it trades completeness against latency.
• Allowed lateness sets how long window state is kept for corrections; beyond it, route late events to a side output rather than dropping silently.
• Triggers control when results are emitted, accumulation mode controls whether a re-emission replaces or adds.
• An idle source partition can freeze the global watermark and stall all output.

🧪 Practice

1. Given events arriving at event times 10, 12, 11, 21, 14 with a 5-second out-of-orderness bound, state the watermark after each and which windows close.
2. Choose an allowed lateness for mobile analytics where phones can be offline for hours, and describe the memory consequence.
3. Interview: Your streaming counts disagree with the nightly batch counts by 0.3%. What is the most likely explanation? (Hint: think about which events the batch job can see that the stream had already decided were too late.)

Lambda and Kappa Architectures

Streaming gives you fresh answers; batch gives you complete, easily corrected ones. For years no single system did both well, and the architectural patterns that emerged are still the clearest way to reason about the trade-off — even where modern engines have narrowed the gap.

Lambda architecture runs both. A batch layer recomputes authoritative views from the full immutable dataset on a schedule; a speed layer computes approximate views over recent events with low latency; a serving layer merges them, using batch results wherever they exist and the speed layer only for the window the batch job has not yet covered. The batch layer's recomputation is a self-healing property: any error in the speed layer is erased by the next batch run.

The cost is the reason Lambda fell out of favour: every piece of business logic exists twice, in two languages, on two engines, maintained by people who must keep them semantically identical. They drift. The drift shows up as numbers that disagree at the boundary, and debugging it means reasoning about two systems at once.

Kappa architecture removes the batch layer. There is one streaming pipeline and one implementation of the logic; the durable, replayable log is the source of truth. "Batch" becomes a special case — reprocessing means replaying the log from the beginning through a new version of the job, writing to a new output table, and switching readers when it catches up. This is the same blue-green idea as deployment (12.3), applied to derived data.

 LAMBDA                                  KAPPA

  events                                  events
    |                                       |
    +--> [ batch layer ]                    v
    |      full recompute            +--------------+
    |      hours, exact              | durable log  |  <- retained long enough
    |         |                      | (Kafka etc.) |     to replay everything
    |         v                      +------+-------+
    |    [ batch views ]                    |
    |         |                       +-----+------+
    +--> [ speed layer ] --+          |            |
           seconds,        |    [ job v1 ]    [ job v2 ]   <- reprocessing is
           approximate     |    (serving)     (replay)        just a second
              |            |          |            |         instance
              v            v          v            v
         [ serving layer: merge ]  table_v1     table_v2
              |                          \        /
              v                       switch readers when v2 catches up
          consumers                          |
                                             v
  Two codebases, one truth.            One codebase, one truth.

def serving_read(key: str, batch_view: dict, speed_view: dict,
                 batch_watermark_ts: int) -> dict:
    """Lambda serving layer: batch results are authoritative up to its
    watermark; the speed layer covers only the gap after it."""
    base = batch_view.get(key, {"count": 0, "as_of": batch_watermark_ts})
    delta = sum(v["count"] for v in speed_view.get(key, [])
                if v["ts"] > batch_watermark_ts)    # avoid double counting
    return {"count": base["count"] + delta, "exact_through": batch_watermark_ts}

def kappa_reprocess(log, new_job, new_table, live_table):
    """Kappa reprocessing: replay from offset 0 into a fresh table, then switch."""
    for record in log.replay(from_offset=0):
        new_job.process(record, sink=new_table)
    if new_table.lag_seconds() < 60:        # caught up with the live pipeline
        return promote(new_table, replaces=live_table)   # atomic pointer swap
    return "still catching up"

Concern	Lambda	Kappa
Codebases for logic	Two	One
Correcting a bug	Next batch run heals it	Replay the log into a new table
Storage requirement	Full historical dataset	Log retention long enough to replay
Reprocessing cost	Routine, already scheduled	A burst of compute on demand
Late data	Batch layer absorbs it	Watermarks and allowed lateness
Operational burden	Two systems in sync	One system, longer retention

Kappa is the better default when the log can retain enough history and the logic is expressible in streaming SQL. Lambda-shaped designs persist where the batch computation is genuinely different in kind — a model retrain, a graph algorithm, a full-table reconciliation against an external system of record. The lakehouse (14.1.3) has softened the distinction further: one table can be appended to by a stream and rewritten by a batch job under the same transaction log, so the "two layers" become two writers to one destination rather than two serving paths.

Key Takeaways

• Lambda runs batch and streaming layers in parallel and merges them; batch recomputation self-heals speed-layer errors.
• Lambda's fatal cost is duplicated business logic that inevitably drifts.
• Kappa keeps one streaming implementation and treats reprocessing as replaying the log into a new table, then switching readers.
• Kappa's prerequisite is log retention long enough to replay the history you care about.
• Lakehouse table formats blur the split: one table, two writers, one transaction log.

🧪 Practice

1. Take a metric computed both in a nightly job and a live dashboard, and list every place the two implementations could diverge.
2. Compute the log retention and storage needed to replay 90 days at 50k events/sec and 400 bytes per event.
3. Interview: You found a bug in a streaming aggregation that has been running for a month. How do you correct the historical output? (Hint: think about what you would need to have retained, and how you would cut consumers over without a period of visibly wrong numbers.)

<a id="143-search-and-recommendation-systems"></a>

14.3 Search and Recommendation Systems

Search and recommendation are the same system viewed from two ends: both reduce millions of candidate items to a short ordered list, under a latency budget, using signals computed elsewhere. This subchapter covers how documents get into an index, how ranking is served under that budget, the multi-stage funnel that makes the scale tractable, the feature infrastructure both stages depend on, how personalization stays fresh, and the experimentation platform without which none of it can be shown to work.

Indexing Pipelines

A search engine cannot scan every document per query, so it inverts the problem ahead of time. An inverted index maps each term to a posting list of the documents containing it, so answering "invoices about refunds" becomes intersecting two short lists instead of reading every document. Everything in search performance follows from that inversion — and from keeping it current.

The indexing pipeline is what turns source records into that structure. Documents are ingested (bulk backfill or incremental change events), normalized (lower casing, Unicode folding, HTML stripping), analyzed into tokens (tokenization, stemming or lemmatization, stop-word handling, n-grams for partial matching), enriched with data the ranker will need (category, popularity, embeddings), and finally written into index segments. Segments are immutable: new documents go into new small segments, deletions are recorded as tombstones, and background merges compact many small segments into fewer large ones. Immutability is what lets readers run lock-free while writers append.

 INDEXING PIPELINE

  source DB / event stream
        |  CDC or bulk export
        v
  [ normalize ] -> [ analyze ] -> [ enrich ] -> [ build segment ] -> [ commit ]
   lowercase       tokenize        category,      immutable file      visible
   strip HTML      stem            popularity,    + tombstones        to readers
   fold Unicode    stop-words      embedding

 INVERTED INDEX (term -> postings)

   "refund"  -> [ doc3(tf=2), doc7(tf=1), doc9(tf=5) ]
   "invoice" -> [ doc1(tf=1), doc3(tf=3), doc9(tf=1) ]

   query "refund invoice" = intersect postings -> {doc3, doc9}
   Postings are sorted by doc id and delta-compressed, so intersection is a
   linear merge over the SHORTEST list first.

from collections import defaultdict

STOP = {"the", "a", "of", "and"}

def analyze(text: str) -> list[str]:
    """Analysis must be IDENTICAL at index and query time. A mismatch here is
    the single most common cause of 'why does this document not match?'."""
    tokens = [t.strip(".,!?").lower() for t in text.split()]
    tokens = [t for t in tokens if t and t not in STOP]
    return [t[:-1] if t.endswith("s") and len(t) > 3 else t for t in tokens]

class Index:
    def __init__(self):
        self.postings: dict[str, dict[int, int]] = defaultdict(dict)
        self.deleted: set[int] = set()      # tombstones; purged at merge time
        self.doc_len: dict[int, int] = {}

    def add(self, doc_id: int, text: str) -> None:
        terms = analyze(text)
        self.doc_len[doc_id] = len(terms)
        for t in terms:
            self.postings[t][doc_id] = self.postings[t].get(doc_id, 0) + 1

    def delete(self, doc_id: int) -> None:
        self.deleted.add(doc_id)            # cheap: no segment rewrite

    def search(self, query: str) -> list[int]:
        lists = [self.postings.get(t, {}) for t in analyze(query)]
        if not lists or any(not l for l in lists):
            return []
        lists.sort(key=len)                 # start from the shortest list
        hits = set(lists[0])
        for l in lists[1:]:
            hits &= l.keys()                # intersect, shrinking as we go
        return sorted(hits - self.deleted)

idx = Index()
idx.add(1, "The invoice was paid")
idx.add(3, "Refunds and invoices for the refund request")
idx.add(9, "Refund policy of the invoice")
print(idx.search("refund invoice"))          # [3, 9]

The operationally interesting property is freshness, and it is bought at a cost. Committing a segment makes it visible but is expensive, so engines batch commits — a refresh interval of one second is typical, which means a document written now is searchable roughly a second later, not instantly. Systems that need true read-your-writes usually overlay a small in-memory buffer of very recent changes on top of the index at query time.

Two more realities shape production indexing. Reindexing is routine: any change to the analyzer, the schema, or an enrichment field requires rebuilding, which is done by building a new index alongside the old one and switching an alias atomically once it is verified — never by mutating in place. And partial updates are a lie in most engines: changing one field re-indexes the whole document, so put volatile signals (view counts, prices) in a separate store the ranker reads at query time rather than inside the document.

Key Takeaways

• An inverted index maps terms to posting lists, turning search into a list intersection instead of a scan.
• Segments are immutable; writes append and deletes tombstone, with background merges compacting them.
• Index-time and query-time analysis must match exactly, or documents silently fail to match.
• Freshness is a commit-interval trade-off, not a free property; overlay recent writes if you need read-your-writes.
• Reindex into a new index and swap an alias; never mutate an index in place.
• Keep volatile signals out of the document — they force full re-indexing.

🧪 Practice

1. Build an inverted index by hand for five short documents and execute a two-term query against it, starting from the shorter posting list.
2. Describe the migration steps for changing the stemming rules on a live index with zero downtime.
3. Interview: A newly created record is not searchable for a few seconds. Is this a bug? (Hint: think about what makes a segment visible to readers, and what you would give up to make it instant.)

Ranking and Relevance Serving

Retrieval decides which documents can match; ranking decides which ones the user sees. Since almost nobody looks past the first few results, ranking quality is effectively the product. The engineering problem is that ranking must be both good and fast: a scoring function that considers a hundred signals cannot be run over a million matching documents inside a 150 ms budget.

Classical lexical ranking uses BM25, a refinement of TF-IDF. Its intuition is worth internalizing because it explains most "why did this rank first?" questions. A term matters more when it appears often in a document (term frequency, with diminishing returns — the tenth occurrence adds little), less when it appears in many documents (inverse document frequency — "the" discriminates nothing), and document length is normalized so a long document does not win merely by containing more words.

import math

def bm25(tf: int, doc_len: int, avg_len: float, n_docs: int, n_with_term: int,
         k1: float = 1.2, b: float = 0.75) -> float:
    """k1 controls how fast term-frequency saturates; b controls how strongly
    length is normalized (b=0 -> no normalization, b=1 -> full)."""
    idf = math.log(1 + (n_docs - n_with_term + 0.5) / (n_with_term + 0.5))
    norm = tf * (k1 + 1) / (tf + k1 * (1 - b + b * doc_len / avg_len))
    return idf * norm

# Same term frequency, different rarity: the rare term dominates the score.
print(round(bm25(tf=3, doc_len=300, avg_len=250, n_docs=1e6, n_with_term=900_000), 3))
print(round(bm25(tf=3, doc_len=300, avg_len=250, n_docs=1e6, n_with_term=120), 3))
# 0.115   <- common term contributes almost nothing
# 8.203   <- rare term carries the match

Lexical scoring fails on vocabulary mismatch: a query for "laptop" does not match a document saying "notebook computer". Semantic (vector) retrieval embeds query and documents into a shared space where meaning, not spelling, determines proximity (14.4.5). Neither approach dominates — lexical is precise on exact identifiers, part numbers, and names; semantic handles paraphrase and intent — so production systems run hybrid retrieval and fuse the two result sets. The standard fusion technique is reciprocal rank fusion, which combines by rank rather than by score and therefore needs no score calibration between systems.

def reciprocal_rank_fusion(rankings: list[list[str]], k: int = 60) -> list[str]:
    """Fuse ranked lists from lexical and vector retrieval.
    Uses only positions, so incomparable score scales do not matter."""
    scores: dict[str, float] = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank)
    return sorted(scores, key=scores.get, reverse=True)

lexical = ["d9", "d3", "d7", "d1"]
vector  = ["d3", "d5", "d9", "d2"]
print(reciprocal_rank_fusion([lexical, vector])[:4])   # ['d3', 'd9', 'd7', 'd5']

Serving architecture follows from the index being too big for one machine. The index is sharded by document, each shard is replicated for throughput and availability, and a coordinator fans out to one replica per shard, merges the top-k from each, and returns the global top-k. This is a textbook fan-out, which means it inherits tail-latency amplification (13.1.2): the query is as slow as its slowest shard.

 SEARCH SERVING PATH (budget: 150 ms p95)

  query -> [ understanding: spellcheck, synonyms, intent ]     ~10 ms
              |
              v  fan-out to every shard, one replica each
        +-----+-----+-----+-----+
        | s0  | s1  | s2  | s3  |   each returns its LOCAL top-100     ~40 ms
        +-----+-----+-----+-----+   (cheap BM25 / ANN scoring)
              |
              v  coordinator merges 400 -> global top-100
        [ re-rank top-100 with the expensive model ]           ~60 ms
              |
              v
        [ blend, dedupe, apply business rules, paginate ]      ~15 ms

  Tail control: hedge the slowest shard, and enforce a per-shard deadline that
  returns partial results rather than failing the whole query.

Two production disciplines round this out. Relevance is measured, not asserted: offline with graded judgements (NDCG, MRR, recall@k) over a fixed query set, and online with engagement metrics via experiments (14.3.6) — because offline gains routinely fail to reproduce online. And business rules belong after ranking, as an explicit blending or filtering stage — pinning, freshness boosts, inventory filters, and diversity constraints applied post-hoc keep the learned model's behaviour understandable rather than entangled with policy.

Key Takeaways

• Retrieval selects candidates, ranking orders them; ranking quality is what users perceive as the product.
• BM25 balances term frequency saturation, inverse document frequency, and length normalization.
• Lexical and semantic retrieval fail differently — hybrid retrieval with reciprocal rank fusion combines them without score calibration.
• Sharded fan-out inherits tail amplification: use per-shard deadlines, partial results, and hedging.
• Measure relevance offline with NDCG-style metrics and confirm online with experiments; offline wins often do not transfer.
• Keep business rules in an explicit post-ranking stage, not inside the model.

🧪 Practice

1. Compute BM25 contributions for a common and a rare term at equal term frequency, and explain the ordering that results.
2. Design the fan-out path for a 40-shard index with a 150 ms budget, stating the per-shard deadline and the partial-result policy.
3. Interview: Users complain that searching an exact product code returns unrelated items. What went wrong? (Hint: think about which retrieval mode is good at exact tokens, and what an analyzer or an embedding does to a string like XR-4410-B.)

Candidate Generation and Re-Ranking

Recommendation faces a harder version of search's problem: there is no query text, and the catalogue may hold hundreds of millions of items. Scoring every item with a good model for every request is off by several orders of magnitude — a 1 ms model over 100M items is 100,000 seconds per request. The universal solution is a funnel: a few stages, each cheaper per item than the next and each narrowing the set, so expensive computation is spent only where it changes the outcome.

Candidate generation (retrieval) cuts 100M items to a few hundred or thousand using methods that are sublinear in catalogue size: approximate nearest neighbour lookup over embeddings, co-visitation and collaborative-filtering tables built offline, inverted-index lookups on attributes, plus deliberate sources like trending, recently viewed, and editorially curated items. Multiple generators run in parallel and their outputs are unioned — each covers a different failure mode, and diversity of sources is what keeps recommendations from collapsing onto one behaviour.

Ranking scores those few hundred candidates with a real model over rich features (user history, item attributes, context, cross features). Re-ranking then adjusts the ordered list for properties that only exist at the list level: diversity (not five items from one seller), business constraints (inventory, margin, contractual promotions), de-duplication, and exploration slots for items with little feedback.

 THE RECOMMENDATION FUNNEL

  catalogue        100,000,000 items
       |  candidate generation: ANN + co-visitation + trending + recent
       |  cost per item: microseconds, sublinear in catalogue size
       v
  candidates             1,000
       |  ranking: gradient-boosted trees or a neural model,
       |  ~100 features per item, ~1 ms per batch of 1,000
       v
  ranked                   200
       |  re-ranking: diversity, business rules, dedupe, exploration
       |  operates on the LIST, not on items independently
       v
  final                     20  -> rendered

  RECALL is set here ---^ (an item missed in generation can never be shown,
  no matter how good the ranker is). PRECISION is set by ranking.

from dataclasses import dataclass

@dataclass
class Item:
    id: str
    score: float
    seller: str
    category: str

def maximal_marginal_relevance(items: list[Item], k: int,
                               lambda_: float = 0.7) -> list[Item]:
    """Balance relevance against redundancy. lambda_=1 is pure relevance;
    lower values force diversity by penalising similarity to what is already
    selected."""
    selected: list[Item] = []
    pool = sorted(items, key=lambda i: -i.score)
    while pool and len(selected) < k:
        best, best_val = None, float("-inf")
        for cand in pool:
            sim = max((similarity(cand, s) for s in selected), default=0.0)
            value = lambda_ * cand.score - (1 - lambda_) * sim
            if value > best_val:
                best, best_val = cand, value
        selected.append(best)
        pool.remove(best)
    return selected

def similarity(a: Item, b: Item) -> float:
    return 0.6 * (a.seller == b.seller) + 0.4 * (a.category == b.category)

cands = [Item("i1", 0.95, "acme", "shoes"), Item("i2", 0.94, "acme", "shoes"),
         Item("i3", 0.93, "acme", "shoes"), Item("i4", 0.80, "globex", "bags"),
         Item("i5", 0.78, "initech", "hats")]
print([i.id for i in maximal_marginal_relevance(cands, k=3)])
# ['i1', 'i4', 'i5'] — pure relevance would have returned three acme shoes

Two structural points make the funnel work in practice. First, recall is determined at the top and can never be recovered: an item not generated cannot be ranked, so candidate generation should over-generate relative to intuition and be monitored with recall@k against what the ranker eventually liked. Second, the funnel needs exploration. A system trained only on what it previously showed enters a feedback loop where unshown items never accumulate the engagement data that would let them be shown; reserving a small fraction of slots for under-explored items (or using a bandit policy) is what keeps the catalogue from ossifying.

The cold-start problem sits in the same place. For new items, generation must fall back on content-based signals (attributes, text and image embeddings) rather than interaction history; for new users, on context — locale, device, referrer, the first few in-session actions — with popularity as the base case. A recommender without an explicit cold-start path silently ignores everything recently added.

Key Takeaways

• Multi-stage funnels make recommendation tractable: cheap generation narrows, expensive ranking orders, list-level re-ranking adjusts.
• Candidate generation sets the recall ceiling — a missed item can never be recovered downstream.
• Run several generators in parallel; each covers a different failure mode.
• Re-ranking handles properties that only exist across a list: diversity, business rules, dedupe.
• Without exploration slots, the system trains on its own output and the catalogue ossifies.
• Cold start needs an explicit content-based and contextual path.

🧪 Practice

1. Size a funnel for a 10M-item catalogue under a 100 ms budget: choose stage widths and the per-item cost each stage can afford.
2. Implement a diversity constraint that caps items per seller in the top 10, and describe how it changes the observed click-through rate.
3. Interview: Recommendations feel repetitive and never surface new items. Diagnose it. (Hint: think about where new items would have to enter the funnel, and what data they lack that the ranker depends on.)

Feature Stores

The subtle failure in production ML is not a bad model; it is a good model fed slightly different data at serving time than at training time. Training reads avg_order_value_30d from a warehouse table computed with a nightly SQL job; serving computes it in application code from a Redis counter. The two definitions diverge by a boundary condition, and model quality drops in a way no test catches. This is training-serving skew, and a feature store exists to make it structurally impossible.

A feature store provides one definition of each feature, materialized into two stores that share it. The offline store (warehouse or lakehouse) holds full history and serves training set construction; the online store (a low-latency key-value store) holds only the current value per entity and serves inference in single-digit milliseconds. One pipeline writes both, so both agree by construction.

 FEATURE STORE

     raw events / tables
            |
      [ ONE feature definition: name, entity, transformation, TTL ]
            |
      +-----+-----------------------------+
      |                                   |
      v                                   v
  OFFLINE STORE                      ONLINE STORE
  (warehouse: full history,          (Redis/DynamoDB: latest value
   point-in-time correct)             per entity, ms reads)
      |                                   |
      v                                   v
  training set construction          inference: get_online_features(user_42)
      |                                   |
      +---------> SAME VALUES <-----------+   <- the whole point

The hardest correctness property a feature store provides is point-in-time correctness. When building a training row for an event at time T, every feature must carry the value it had at T — not the value it has now. Joining on the entity key alone leaks future information into training, the model looks excellent offline, and it collapses in production. This is label leakage, and it is the most expensive silent bug in applied ML.

from bisect import bisect_right
from dataclasses import dataclass

@dataclass
class FeatureValue:
    ts: int
    value: float

def point_in_time_lookup(history: list[FeatureValue], event_ts: int,
                         max_staleness: int | None = None) -> float | None:
    """Return the feature value KNOWN AT event_ts — never a later one.
    `history` must be sorted by ts."""
    times = [f.ts for f in history]
    i = bisect_right(times, event_ts) - 1         # last value at or before T
    if i < 0:
        return None                               # feature did not exist yet
    fv = history[i]
    if max_staleness is not None and event_ts - fv.ts > max_staleness:
        return None                               # too old to be meaningful
    return fv.value

hist = [FeatureValue(100, 1.0), FeatureValue(200, 2.0), FeatureValue(300, 3.0)]
print(point_in_time_lookup(hist, event_ts=250))   # 2.0  correct
print(hist[-1].value)                             # 3.0  <- the leaky shortcut:
# using the latest value trains the model on information from the future.

Features are typically grouped by how they are computed: batch features (aggregations over days of history, recomputed on a schedule), streaming features (windowed aggregates updated within seconds), and on-demand features (computed at request time from the request itself — device, time of day, the current cart). The store should also hold metadata that makes features reusable: owner, freshness SLA, description, and which models consume it, so that deprecating a feature is a lookup rather than an archaeology project.

Whether to adopt a dedicated feature store platform is a scale question. With one model and three features, a shared SQL definition plus a documented online write path is sufficient. The platform pays for itself when many models share features, when point-in-time joins are being written by hand repeatedly, or when nobody can say which pipeline produces a given value.

Key Takeaways

• A feature store's purpose is one definition of each feature serving both training and inference, eliminating training-serving skew.
• Offline store = full history for training; online store = latest value for millisecond inference; one pipeline writes both.
• Point-in-time correctness is mandatory: joining on the entity key alone leaks future information and inflates offline metrics.
• Features divide into batch, streaming, and on-demand, with different freshness guarantees.
• Metadata — owner, freshness SLA, consumers — is what makes features reusable and safely deprecable.

🧪 Practice

1. Write the offline and online definitions for "number of purchases in the last 7 days" and identify where the two could drift.
2. Construct a training row with a point-in-time join and show numerically what a naive latest-value join would leak.
3. Interview: A model scores 0.92 AUC offline and barely beats random online. What do you check first? (Hint: think about whether any feature in training could only have been known after the label was determined.)

Real-Time Personalization

Batch personalization computes recommendations nightly and serves them all day. That is adequate for stable preferences and useless for intent, which changes within a session: a user who has viewed three tents in five minutes wants camping gear now, not the reading recommendations computed at 3 a.m. Real-time personalization closes that loop, and the engineering question is which parts of the loop must be fast — because making all of it fast is unnecessary and expensive.

The workable decomposition splits the system by rate of change. Expensive, slow-moving artefacts — model weights, item embeddings, co-visitation tables — are computed offline. Fast-moving state — the events of the current session — is maintained online and combined with those artefacts at request time. The model does not need retraining to react; it needs the session's features to be current.

 TWO LOOPS AT DIFFERENT SPEEDS

  SLOW LOOP (hours/days)              FAST LOOP (seconds)
  ---------------------               -------------------
  event history --> training          click/view/add-to-cart
        |                                     |
        v                                     v  <100 ms
  model weights, item embeddings,      [ stream processor ]
  co-visitation tables                        |  session aggregates
        |                                     v
        +--------------+           [ online store: user:42 -> {viewed: [...],
                       |             session_intent: "camping", cart: [...] } ]
                       v                      |
                [ inference at request time ] <+
                       |  slow artefacts x fresh session state
                       v
                  personalized results

import time
from collections import deque

class SessionState:
    """Online session features: bounded, TTL'd, and cheap to update.
    Kept in the online store keyed by user or device, with a short TTL so an
    abandoned session does not personalize tomorrow's visit."""

    MAX_EVENTS = 50
    TTL_S = 30 * 60

    def __init__(self):
        self.events: deque = deque(maxlen=self.MAX_EVENTS)   # bounded memory
        self.last_seen = time.time()

    def record(self, event_type: str, item_id: str, category: str) -> None:
        now = time.time()
        if now - self.last_seen > self.TTL_S:
            self.events.clear()                  # stale session: start fresh
        self.events.append((now, event_type, item_id, category))
        self.last_seen = now

    def features(self) -> dict:
        weights = {"view": 1.0, "add_to_cart": 4.0, "purchase": 8.0}
        by_cat: dict[str, float] = {}
        now = time.time()
        for ts, etype, _item, cat in self.events:
            # Exponential recency decay: 5-minute half-life, so intent from
            # 20 minutes ago counts far less than the last two clicks.
            decay = 0.5 ** ((now - ts) / 300)
            by_cat[cat] = by_cat.get(cat, 0.0) + weights.get(etype, 1.0) * decay
        top = max(by_cat, key=by_cat.get) if by_cat else None
        return {"session_intent": top,
                "intent_strength": round(by_cat.get(top, 0.0), 2) if top else 0.0,
                "recent_items": [e[2] for e in list(self.events)[-5:]]}

s = SessionState()
for etype, item, cat in [("view", "t1", "camping"), ("view", "t2", "camping"),
                         ("add_to_cart", "t2", "camping"), ("view", "b9", "books")]:
    s.record(etype, item, cat)
print(s.features())
# {'session_intent': 'camping', 'intent_strength': 6.0, 'recent_items': [...]}

Three constraints dominate the design. Latency: personalization sits on the request path, so the online feature read plus scoring must fit the page budget (13.1.1) — one round trip to the online store, batched, not one per candidate. Cost: recomputing everything per request is unaffordable, so cache the expensive parts keyed by (user, context) with a short TTL and invalidate on significant events. Graceful degradation: if the online store is slow or the session is empty, fall back to segment-level or popularity-based results rather than failing the page — personalization is an enhancement, and a page that fails to render because a recommender timed out is a worse outcome than a generic one.

Finally, real-time personalization amplifies feedback loops. Reacting instantly to the last click can trap a user in a narrow slice of the catalogue — the "I bought one fridge, now everything is fridges" failure. Counter it deliberately: decay intent over time (as above), suppress items already purchased, blend session signals with longer-term preferences, and keep the exploration slots from 14.3.3.

Key Takeaways

• Split the system by rate of change: train models offline, keep session state online, and combine them at request time.
• Freshness of features, not of model weights, is what makes personalization feel responsive.
• Bound session state with a size cap and a TTL, and decay signals by recency.
• Budget one batched online-store read on the request path, and cache expensive intermediate results with short TTLs.
• Always degrade to segment or popularity results rather than failing the page.
• Instant reaction narrows the catalogue: decay intent, suppress purchased items, and keep exploring.

🧪 Practice

1. Define the session features you would maintain for a travel booking site, with their TTLs and decay behaviour.
2. Design the fallback chain for the recommendation widget when the online store exceeds its deadline.
3. Interview: A user buys a washing machine and now sees only washing machines. How do you fix this? (Hint: think about which post-purchase signal should suppress a category, and how long that suppression should last.)

A/B Testing Infrastructure

Every claim in the previous five topics — a better ranker, fresher features, more diversity — is a hypothesis until it is measured on real users. Offline metrics routinely disagree with online outcomes, because offline evaluation cannot model how users respond to a changed interface. A/B testing infrastructure is the system that turns opinions into measured deltas, and it is as much a piece of production engineering as the ranker itself.

The core mechanism is random assignment that is deterministic and sticky: the same user must get the same variant on every request, across devices and sessions, without a database lookup on the request path. Hashing the unit ID with the experiment ID gives exactly that — stable, uniform, independent across experiments, and computable in microseconds.

import hashlib

def assign(unit_id: str, experiment_id: str, variants: dict[str, float]) -> str:
    """Deterministic, sticky, salted-per-experiment bucketing.
    Salting by experiment_id makes assignments INDEPENDENT across experiments —
    without it, the same users always land in the same bucket everywhere and
    experiments become correlated."""
    digest = hashlib.sha256(f"{experiment_id}:{unit_id}".encode()).hexdigest()
    point = int(digest[:16], 16) / 2 ** 64      # uniform in [0, 1)
    cumulative = 0.0
    for name, share in variants.items():
        cumulative += share
        if point < cumulative:
            return name
    return "control"

variants = {"control": 0.5, "treatment": 0.5}
counts = {"control": 0, "treatment": 0}
for i in range(100_000):
    counts[assign(f"user-{i}", "ranker_v4", variants)] += 1
print(counts)                       # ~50/50, and stable across restarts

# The same user gets DIFFERENT buckets in different experiments:
print(assign("user-7", "ranker_v4", variants),
      assign("user-7", "checkout_copy", variants))

The design decisions that matter most are usually not statistical. The randomization unit must match the level at which the effect and the user experience occur: user-level for anything visible in the interface, session-level only for effects that cannot leak across sessions, and cluster-level (household, city, market) when treated users interact with untreated ones — a marketplace change affects both sides, and user-level randomization then violates the independence the statistics assume. Guardrail metrics — latency, error rate, crash rate, revenue — must be evaluated alongside the target metric so a variant that raises clicks by slowing the page is caught rather than shipped.

 EXPERIMENT LIFECYCLE

  hypothesis + primary metric + MDE + guardrails, written down BEFORE launch
        |
        v
  [ A/A test ] validate the pipeline: two identical variants must show no
        |      significant difference (if they do, assignment or logging is broken)
        v
  [ ramp: 1% -> 5% -> 50% ] each step gated on guardrails, not on the win
        |
        v
  [ run to the pre-computed sample size ] <- no peeking-and-stopping
        |
        v
  [ analyse: primary + guardrails + key segments ] -> ship / iterate / kill

import math

def sample_size_per_arm(baseline_rate: float, mde_relative: float,
                        alpha: float = 0.05, power: float = 0.8) -> int:
    """How many units per arm are needed to detect a relative lift of
    `mde_relative`. Compute this BEFORE launching — it tells you whether the
    experiment is even feasible with your traffic."""
    z_a, z_b = 1.96, 0.84                      # two-sided alpha=0.05, power=0.8
    p1 = baseline_rate
    p2 = baseline_rate * (1 + mde_relative)
    p_bar = (p1 + p2) / 2
    num = (z_a * math.sqrt(2 * p_bar * (1 - p_bar))
           + z_b * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2
    return math.ceil(num / (p2 - p1) ** 2)

print(sample_size_per_arm(0.10, 0.05))    # 5% lift on a 10% rate: ~ 61,000/arm
print(sample_size_per_arm(0.10, 0.01))    # 1% lift: ~1,500,000/arm
# Small effects are expensive to detect. If the traffic is not there, the
# honest conclusion is that the experiment cannot answer the question.

Three failure modes are worth naming because they recur. Peeking: checking results continuously and stopping at the first significant reading inflates the false-positive rate far above the nominal 5% — fix it with a pre-computed sample size, or use sequential testing designed for continuous monitoring. Sample ratio mismatch: if a 50/50 split delivers 50.4/49.6 on millions of users, something in assignment, logging, or filtering is broken and the result is untrustworthy regardless of how good it looks; SRM should be an automatic, blocking check. Multiple comparisons: testing twenty metrics guarantees one "significant" result by chance, which is why a single primary metric is declared in advance and the rest are treated as diagnostics.

Key Takeaways

• Assignment must be deterministic, sticky, and salted per experiment so experiments stay independent.
• Match the randomization unit to where the effect occurs; use cluster randomization when treated and untreated users interact.
• Compute the required sample size before launch — if traffic cannot detect the effect, the experiment cannot answer the question.
• Run an A/A test to validate the pipeline, and make sample-ratio mismatch a blocking check.
• Declare one primary metric in advance and always evaluate guardrails (latency, errors, revenue) alongside it.
• Peeking inflates false positives; ramp for safety, but analyse at the planned sample size.

🧪 Practice

1. Implement bucketing for a 3-arm experiment with 10/10/80 traffic and verify the distribution empirically.
2. Compute the sample size needed to detect a 2% relative lift on a 4% baseline conversion rate, then state how many days your traffic would need.
3. Interview: A 50/50 experiment shows 51.2% of users in treatment. Do you trust the results? (Hint: think about how large that deviation is relative to random variation at your sample size, and what kinds of bugs produce a one-sided loss of units.)

<a id="144-ml-and-ai-serving-systems"></a>

14.4 ML and AI Serving Systems

A trained model is not a product; the infrastructure that runs it under a latency budget, ships new versions safely, and notices when it quietly stops working is. This subchapter covers how training and inference infrastructure differ, when to precompute predictions versus serve them live, how models are versioned and rolled out, how expensive accelerators are scheduled and batched, the retrieval pipelines behind vector search and RAG, and the monitoring that catches degradation nobody reported.

Model Training vs. Inference Infrastructure

Training and inference are both "running a model", which hides the fact that they are almost opposite workloads. Training is a batch job: it consumes an entire dataset repeatedly, runs for hours or weeks, saturates hardware deliberately, and can be restarted from a checkpoint if a machine dies. Inference is an online service: it handles one small request at a time, has a latency SLO, scales with traffic rather than data, and every failure is user-visible. Designing one infrastructure for both produces a system that is bad at each.

The differences propagate into every operational choice. Training benefits from the cheapest possible capacity — spot or preemptible instances — because checkpointing makes interruption survivable; inference cannot use interruptible capacity for its baseline. Training scales by adding accelerators to one job with data or model parallelism; inference scales by adding independent replicas behind a load balancer, which is an ordinary stateless-service problem (4.x). Training failures cost time; inference failures cost requests.

 TRAINING                              INFERENCE

  full dataset (TB)                     one request (KB)
        |                                     |
  [ distributed job, N accelerators ]   [ replica pool, autoscaled by RPS ]
        |  checkpoint every N steps           |  p99 latency SLO
        v                                     v
  model artifact + metrics              prediction (ms)

  optimise for: throughput, cost        optimise for: tail latency, availability
  hardware: many accelerators,          hardware: right-sized accelerator or CPU,
    interruptible OK                      always-on, no preemption
  failure: resume from checkpoint       failure: shed load, fall back, degrade
  data:    sequential bulk reads        data:  point lookups (feature store)

The critical shared artefact is the model registry: a versioned store holding each model binary alongside the metadata that makes it deployable and auditable — training data snapshot, code commit, hyperparameters, evaluation metrics, and approval stage. Without it, "which model is in production and what was it trained on?" becomes unanswerable, which makes both rollback and incident analysis impossible.

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class ModelVersion:
    name: str
    version: str
    artifact_uri: str
    framework: str
    training_data_snapshot: str      # exact table version / date range
    code_commit: str                 # reproducibility: what produced this
    metrics: dict[str, float]
    stage: str = "staging"           # staging -> production -> archived
    created_at: datetime = field(default_factory=datetime.utcnow)

class Registry:
    def __init__(self, gates: dict[str, float]):
        self.versions: list[ModelVersion] = []
        self.gates = gates           # minimum metrics required for production

    def promote(self, mv: ModelVersion) -> str:
        failed = [m for m, threshold in self.gates.items()
                  if mv.metrics.get(m, 0.0) < threshold]
        if failed:
            return f"blocked: {failed} below gate"      # never promote by hand
        for other in self.versions:                     # exactly one production
            if other.name == mv.name and other.stage == "production":
                other.stage = "archived"                # keep it: rollback target
        mv.stage = "production"
        self.versions.append(mv)
        return f"{mv.name}:{mv.version} promoted"

reg = Registry(gates={"auc": 0.78, "calibration_error": -0.05})
print(reg.promote(ModelVersion(
    name="ranker", version="4.2.0", artifact_uri="s3://models/ranker/4.2.0",
    framework="xgboost", training_data_snapshot="events@2026-08-25",
    code_commit="9f3ac21", metrics={"auc": 0.81, "calibration_error": -0.01})))

Between the two sits the training pipeline, which should be as automated as a deployment pipeline: assemble the dataset with point-in-time correct features (14.3.4), train, evaluate against a frozen holdout and against the current production model, register the artifact, and gate promotion on the evaluation. Reproducibility is the property that makes all of this debuggable — pin data snapshots, code commits, and random seeds, because "retrain and see" is not an incident response if each retrain produces a different model.

Key Takeaways

• Training is a throughput-optimised batch job; inference is a latency-optimised online service — do not build one system for both.
• Training tolerates preemptible capacity thanks to checkpointing; inference baselines cannot.
• Training scales by parallelising one job; inference scales by adding stateless replicas.
• A model registry with data snapshot, code commit, metrics, and stage is what makes rollback and audit possible.
• Automate the training pipeline and gate promotion on evaluation, never on manual judgement.
• Pin data, code, and seeds: without reproducibility, debugging a model is guesswork.

🧪 Practice

1. List the metadata you would store for each model version so a two-month-old prediction could be explained.
2. Design capacity for a training job on preemptible instances: checkpoint interval, expected restart cost, and the break-even against on-demand.
3. Interview: A model performed well last quarter and worse now, with no code change. How do you investigate? (Hint: think about what else feeding the model has changed, and what you would need to have recorded to prove it.)

Online vs. Offline Inference

Not every prediction needs to be computed while a user waits. The choice between online (synchronous, on the request path) and offline (batch, precomputed and stored) inference is the single largest lever on cost and latency in an ML system, and it is decided by two questions: does the input exist only at request time, and how many of the possible predictions will actually be read?

Offline inference precomputes predictions on a schedule and writes them to a key-value store; serving is then a lookup, costing a millisecond and no accelerator. It is the right answer when inputs change slowly and the prediction space is enumerable — daily churn scores per customer, nightly recommendation slates, precomputed embeddings. Its limits are freshness (a prediction is as old as the last run) and waste (you pay to score every entity, including the 95% that are never requested).

Online inference computes at request time. It is required whenever the input includes something only known then — the query text, the current cart, the uploaded image — or when the prediction space is effectively infinite. The cost is that model execution now sits inside the latency budget and must be capacity-planned for peak traffic.

 DECIDING

  Does the input exist before the request?
    |                              |
   no                             yes
    |                              |
  ONLINE                    Is the prediction space small enough to enumerate?
                                   |                        |
                                  yes                       no
                                   |                        |
                            How fresh must it be?         ONLINE
                             |            |
                          hours        seconds
                             |            |
                          OFFLINE      STREAMING/NEAR-LINE
                                       (recompute on the triggering event)

 COST SHAPE
   offline:  entities x cost_per_score, paid regardless of reads
   online:   requests x cost_per_score, paid only for what is used

def compare_strategies(entities: int, requests_per_day: int,
                       cost_per_score_ms: float = 8.0,
                       accel_cost_per_hour: float = 3.0):
    """Which is cheaper depends entirely on the read/entity ratio."""
    ms_per_hour = 3_600_000
    offline_ms = entities * cost_per_score_ms          # score everyone, daily
    online_ms  = requests_per_day * cost_per_score_ms  # score only what is asked
    fmt = lambda ms: (ms / ms_per_hour) * accel_cost_per_hour
    return {
        "offline_$/day": round(fmt(offline_ms), 2),
        "online_$/day": round(fmt(online_ms), 2),
        "coverage": f"{min(requests_per_day / entities, 1):.1%} of entities read",
    }

print(compare_strategies(entities=50_000_000, requests_per_day=2_000_000))
# {'offline_$/day': 333.33, 'online_$/day': 13.33, 'coverage': '4.0%'}
# Precomputing 50M scores to serve 2M reads wastes 96% of the compute.

print(compare_strategies(entities=100_000, requests_per_day=40_000_000))
# {'offline_$/day': 0.67, 'online_$/day': 266.67, 'coverage': '100.0%'}
# Here precomputation is ~400x cheaper, and it also removes the latency.

Two hybrids cover most real systems. Near-line (streaming) inference recomputes a prediction when a triggering event arrives rather than on a fixed schedule — the user's slate is refreshed seconds after they add an item to the cart — giving most of offline's cheapness with much of online's freshness. Cache-augmented online inference serves live but caches by a normalized input key; for workloads with repeated inputs (popular queries, identical prompts) a cache hit rate of 40-60% is common and directly halves accelerator spend, provided the cache key includes every input that changes the output, model version included.

Batch inference also has its own operational shape: it is a data pipeline, so it needs the same treatment as 14.1.4 — idempotent writes, partitioned outputs, a freshness SLA, and a quality gate before the new predictions replace the old ones. A silently failed nightly scoring job that leaves yesterday's predictions in place is easy to miss for weeks.

Key Takeaways

• Offline inference precomputes and turns serving into a lookup; online computes per request and is required when the input exists only then.
• The cost comparison hinges on how many precomputed predictions are actually read.
• Near-line inference — recompute on a triggering event — captures most of both benefits.
• Cache online predictions by a normalized key that includes the model version.
• Treat batch inference as a data pipeline: idempotent, partitioned, monitored for freshness.

🧪 Practice

1. For three predictions in a product you know, decide online or offline and justify each with the input-availability and coverage tests.
2. Compute the daily cost of both strategies for 5M entities and 300k daily requests at 15 ms per score.
3. Interview: Precomputed recommendations for 100M users are getting expensive and 90% are never seen. What do you change? (Hint: think about who actually shows up, and what you could compute on their arrival instead.)

Model Versioning and Rollout

A model is a deployable artefact, so everything from 12.3 applies — but with an extra hazard: a bad model does not crash. It returns confident, well-formed, worse predictions. Error-rate and latency dashboards stay green while the product quietly degrades, which is why model rollout needs quality signals in the release gate, not just health checks.

Versioning must therefore cover more than the binary. A model version is the tuple (weights, feature definitions, preprocessing code, and the contract of its input/output schema). Ship the preprocessing with the model — the classic production failure is deploying new weights against the old feature transformation and getting garbage from two individually correct components.

 ROLLOUT LADDER FOR MODELS

  1. SHADOW      real traffic -> both models; only the old model's output is used.
                 Measures latency and prediction distribution with zero user risk.
                 Cannot measure business impact: nobody saw the new predictions.
                        |
  2. CANARY      1-5% of traffic served by the new model. Watch guardrails
                 (latency, errors) AND quality proxies (CTR, conversion).
                        |
  3. A/B TEST    50/50 with a declared primary metric (14.3.6). This is the only
                 stage that proves the model is BETTER, not merely not worse.
                        |
  4. RAMP        50% -> 100%, previous version kept warm and registered.
                        |
  5. ROLLBACK    a config change pointing at the previous version. Seconds,
                 not a redeploy — the old artifact never left the registry.

class ModelRouter:
    """Serve multiple model versions behind one endpoint. Rollback is a
    weight change, not a deployment."""

    def __init__(self, models: dict[str, object], weights: dict[str, float],
                 shadow: str | None = None):
        self.models, self.weights, self.shadow = models, weights, shadow

    def predict(self, request, unit_id: str):
        version = assign_weighted(unit_id, self.weights)   # sticky per user
        result = self.models[version].predict(request)

        if self.shadow:
            # Fire-and-forget: never let the shadow model affect the response
            # path, its latency, or its failure modes.
            try:
                shadow_out = self.models[self.shadow].predict(request)
                log_shadow(request_id=request.id, live=result,
                           shadow=shadow_out, shadow_version=self.shadow)
            except Exception:
                pass                                       # shadow never fails live

        log_prediction(request.id, version, result)        # for later evaluation
        return result, version

def assign_weighted(unit_id: str, weights: dict[str, float]) -> str:
    import hashlib
    h = int(hashlib.sha256(unit_id.encode()).hexdigest()[:16], 16) / 2 ** 64
    acc = 0.0
    for version, w in weights.items():
        acc += w
        if h < acc:
            return version
    return next(iter(weights))

Logging every prediction with its model version and input features is what makes all later analysis possible: attributing a metric change to a version, building the next training set from real traffic, and debugging a specific bad output. Without version-tagged prediction logs, a quality regression is undiagnosable.

Two model-specific rollout concerns deserve care. Sticky assignment matters more than for ordinary deployments, because a user flipping between two rankers sees results that reorder inexplicably between page loads. And retraining is a release: an automated retrain that promotes itself without the ladder above is a direct path to production from a data pipeline, which means any upstream data bug ships straight to users. Automate retraining freely; automate promotion only behind the same gates a human would apply.

Key Takeaways

• A bad model degrades quality without raising errors — release gates need quality signals, not just health checks.
• Version the whole bundle: weights, preprocessing, feature definitions, and schema contract.
• Shadow measures safety, canary limits blast radius, A/B proves improvement.
• Keep every version in the registry so rollback is a routing weight change, not a redeploy.
• Log predictions with model version and features: it is the basis of attribution, debugging, and future training sets.
• Automate retraining, but gate promotion — otherwise upstream data bugs ship themselves.

🧪 Practice

1. Define the promotion gates for a fraud model, including quality thresholds and the guardrails that must not regress.
2. Design a shadow deployment and list the comparisons you would compute between live and shadow outputs.
3. Interview: A new model has better offline metrics but conversion drops in the canary. What do you check? (Hint: think about what the offline holdout could not contain, and about the components deployed alongside the weights.)

GPU Scheduling and Batching

Accelerators change the economics of serving. A GPU is expensive, cannot be oversubscribed the way CPUs can, and — critically — is dramatically more efficient on batched work than on single requests, because its throughput comes from parallel execution units that a batch of one leaves idle. Serving one request at a time on a GPU is like running a freight train with a single parcel: the trip costs the same either way.

Dynamic batching is the standard response: the server holds incoming requests in a queue for a short window and executes them as one batch. This trades a small, bounded latency increase for a large throughput gain, and the tuning is a direct latency/throughput trade — the maximum wait must come out of the latency budget (13.1.1), and the maximum batch size is bounded by accelerator memory.

import asyncio, time

class DynamicBatcher:
    """Accumulate requests up to max_batch or max_wait_ms, whichever first."""

    def __init__(self, model, max_batch: int = 32, max_wait_ms: float = 10):
        self.model, self.max_batch, self.max_wait = model, max_batch, max_wait_ms
        self.queue: list[tuple] = []
        self.lock = asyncio.Lock()

    async def predict(self, x):
        fut = asyncio.get_event_loop().create_future()
        async with self.lock:
            self.queue.append((x, fut))
            first_in_batch = len(self.queue) == 1
            full = len(self.queue) >= self.max_batch
        if full:
            await self._flush()                       # do not wait: batch is full
        elif first_in_batch:
            asyncio.get_event_loop().call_later(       # start the window timer
                self.max_wait / 1000, lambda: asyncio.ensure_future(self._flush()))
        return await fut

    async def _flush(self):
        async with self.lock:
            batch, self.queue = self.queue, []
        if not batch:
            return
        outputs = self.model.predict_batch([x for x, _ in batch])   # one GPU call
        for (_, fut), out in zip(batch, outputs):
            if not fut.done():
                fut.set_result(out)

def throughput(batch_size: int, fixed_ms: float = 6.0, per_item_ms: float = 0.4):
    """Batch cost = fixed overhead + marginal per-item cost. The fixed part is
    why batching pays: it is amortised across the whole batch."""
    latency = fixed_ms + per_item_ms * batch_size
    return {"batch": batch_size, "batch_latency_ms": round(latency, 1),
            "rps": round(batch_size / latency * 1000)}

for b in (1, 8, 32, 128):
    print(throughput(b))
# {'batch': 1,   'batch_latency_ms': 6.4,  'rps': 156}
# {'batch': 8,   'batch_latency_ms': 9.2,  'rps': 870}
# {'batch': 32,  'batch_latency_ms': 18.8, 'rps': 1702}
# {'batch': 128, 'batch_latency_ms': 57.2, 'rps': 2238}
# 11x the throughput for 3x the latency at batch 32 — then diminishing returns.

Utilization is the other half. A dedicated GPU per model wastes capacity whenever that model is idle, so serving platforms multiplex: several models share one accelerator via MPS or time slicing, MIG partitions a large GPU into isolated slices with guaranteed memory, and Kubernetes device plugins expose accelerators as schedulable resources. Model size relative to accelerator memory is what determines which of these is possible.

 SERVING CLUSTER

  requests --> [ router: model name -> pool ]
                    |            |            |
              +-----v----+ +-----v----+ +-----v----+
              | GPU node | | GPU node | | GPU node |
              | A: rank  | | A: rank  | | B: embed |   <- pools sized separately
              | (batched)| | (batched)| | (batched)|      per model's traffic
              +----------+ +----------+ +----------+

  QUEUEING: accelerator pools run at HIGH utilisation by design, and 13.1.4
  says queue delay explodes near saturation. Therefore:
    - admission control: reject beyond a queue depth rather than queue forever
    - a request deadline: drop work whose caller has already timed out
    - separate queues per priority so interactive traffic is not stuck behind
      a bulk backfill

Three further techniques cut cost substantially and are worth knowing by name. Quantization (fp16, int8, and lower) shrinks weights and speeds execution for a small accuracy cost — usually the highest return per unit of effort. Distillation trains a small model to imitate a large one, useful when the large model is affordable offline but not online. Compilation to an optimized runtime (TensorRT, ONNX Runtime, torch.compile) fuses operations and often yields 2-4x with no modelling change. For autoregressive text models, continuous batching goes further than static dynamic batching: because each sequence finishes at a different step, the server admits new sequences into the batch as slots free up rather than waiting for the whole batch to complete, which is what makes high-throughput LLM serving practical alongside KV-cache reuse.

Finally, autoscaling accelerators is slower and coarser than autoscaling CPU pods: instances are scarce, expensive, and take minutes to become ready. Scale on queue depth or batch-fill rate rather than on GPU utilization alone, keep a warm buffer sized to your scale-up time, and put non-urgent work (batch scoring, evaluation runs) on preemptible capacity that can be displaced by interactive traffic.

Key Takeaways

• Accelerators are throughput devices: unbatched serving wastes most of their capacity.
• Dynamic batching trades a bounded wait for large throughput gains; the wait must fit the latency budget.
• Batch cost is fixed overhead plus marginal per-item cost — the fixed part is what batching amortises, with diminishing returns past a point.
• Multiplex models onto accelerators (MPS, MIG, device plugins) instead of dedicating one per model.
• High utilization means queueing: use admission control, deadlines, and priority queues.
• Quantization, distillation, and compilation often beat buying more hardware; continuous batching is what makes LLM serving efficient.

🧪 Practice

1. Given a 40 ms budget and a model at 6 ms fixed plus 0.4 ms per item, find the largest batch size and wait window that fits.
2. Design admission control for an inference service: queue depth limit, deadline handling, and what the client sees on rejection.
3. Interview: GPU utilization reads 30% but latency is high. What is happening? (Hint: think about what a utilization metric measures during a kernel versus between kernels, and where requests might be spending their time instead.)

Vector Search and Retrieval Pipelines

Keyword matching fails on meaning: "how do I cancel my plan" should match a document titled "Ending your subscription", which shares no significant terms. Embeddings solve this by mapping text (or images, or audio) into a high-dimensional vector space where semantic similarity becomes geometric proximity. Retrieval then means finding the nearest vectors to a query vector.

Exact nearest-neighbour search over 100 million vectors is a full scan, so production systems use approximate nearest neighbour (ANN) indexes that trade a little recall for orders of magnitude in speed. Two families dominate. HNSW builds a multi-layer navigable graph and searches by greedy descent — excellent recall and latency, memory-hungry, and awkward to update in bulk. IVF clusters vectors and scans only the nearest few clusters — cheaper memory, tunable via how many clusters are probed — and is usually combined with product quantization, which compresses vectors to a few bytes each so that billions fit in RAM.

 ANN INDEX STRUCTURES

 HNSW (graph)                          IVF + PQ (clustered + compressed)

  layer 2:  o------------o              centroid A   centroid B   centroid C
             \          /                 . . .        . . .        . . .
  layer 1:  o--o----o--o                  . q .        . . .        . . .
             \  \  /  /                    ^
  layer 0:  o-o-o-o-o-o-o  <- all       query lands nearest A and B ->
            vectors, dense links        scan ONLY those lists (nprobe=2),
                                        each vector stored as ~16 compressed
  search: descend layers, greedy        bytes instead of 3072 raw bytes
  moves toward the query
                                        recall knob: nprobe (more = slower,
  recall knob: efSearch                                       better recall)

import math, random

def cosine(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a)); nb = math.sqrt(sum(y * y for y in b))
    return dot / (na * nb + 1e-12)

class IVFIndex:
    """Minimal IVF: assign vectors to centroids, probe only the nearest few."""

    def __init__(self, centroids: list[list[float]]):
        self.centroids = centroids
        self.lists: dict[int, list[tuple[str, list[float]]]] = {
            i: [] for i in range(len(centroids))}

    def add(self, doc_id: str, vec: list[float]) -> None:
        best = max(range(len(self.centroids)),
                   key=lambda i: cosine(vec, self.centroids[i]))
        self.lists[best].append((doc_id, vec))

    def search(self, q: list[float], k: int = 5, nprobe: int = 2):
        # Probe only the nprobe nearest clusters: this is the approximation.
        order = sorted(range(len(self.centroids)),
                       key=lambda i: -cosine(q, self.centroids[i]))
        scanned = [pair for i in order[:nprobe] for pair in self.lists[i]]
        ranked = sorted(((cosine(q, v), d) for d, v in scanned), reverse=True)
        return ranked[:k], len(scanned)

random.seed(7)
dim = 16
cents = [[random.gauss(0, 1) for _ in range(dim)] for _ in range(8)]
idx = IVFIndex(cents)
for i in range(2000):
    idx.add(f"d{i}", [random.gauss(0, 1) for _ in range(dim)])
hits, scanned = idx.search([random.gauss(0, 1) for _ in range(dim)], nprobe=2)
print(f"scanned {scanned} of 2000 vectors -> {[d for _, d in hits]}")

The most common production use is retrieval-augmented generation (RAG): retrieve relevant passages and give them to a language model as context, so answers are grounded in your data rather than in the model's parameters. As a system, RAG is a pipeline whose quality is usually limited by retrieval, not by the model — most disappointing RAG deployments are retrieval problems wearing a generation costume.

 RAG PIPELINE

 INDEXING (offline)
   documents -> chunk (respect structure; overlap ~10-20%)
             -> embed (batch on the accelerator)
             -> store {vector, text, metadata: source, section, updated_at, acl}
             -> reindex when the embedding MODEL changes (vectors are not
                comparable across models — this is a full rebuild)

 QUERY (online)                                   budget
   query -> rewrite/expand                        ~20 ms
         -> hybrid retrieve (ANN + BM25)          ~30 ms   <- 14.3.2 fusion
         -> filter by ACL and freshness           ~5 ms    <- MUST be enforced
         -> re-rank top-50 with a cross-encoder   ~60 ms      here, not in the
         -> assemble context, cite sources        ~5 ms       prompt
         -> generate                              ~1-3 s

Four engineering decisions carry most of the quality. Chunking: chunks too large dilute the embedding and waste context, too small lose the surrounding meaning; split on document structure with modest overlap rather than at a fixed character count. Hybrid retrieval: pure vector search misses exact identifiers, error codes, and names, so fuse with lexical results (14.3.2). Re-ranking: a cross-encoder that reads query and passage together is far more accurate than embedding similarity and is affordable over 50 candidates. Access control: filter by permissions during retrieval — never rely on instructing the model not to reveal something, because a retrieved passage in the context window is already exposed.

Operationally, treat the vector index like any other index (14.3.1): rebuilding after an embedding-model change is a full reindex into a new collection with an alias swap; freshness needs an update path for changed documents; and evaluation needs a fixed query set with judged relevance so you can tell whether a chunking or model change actually helped.

Key Takeaways

• Embeddings turn semantic similarity into geometric proximity; ANN indexes trade recall for speed to make it scale.
• HNSW favours recall and latency at high memory cost; IVF+PQ favours memory and tunes recall via nprobe.
• RAG quality is usually bounded by retrieval, not generation.
• Chunk on structure with overlap, retrieve hybrid, then re-rank the top candidates with a cross-encoder.
• Enforce access control during retrieval — anything in the context is already disclosed.
• Changing the embedding model invalidates every stored vector: plan a full reindex with an alias swap.

🧪 Practice

1. Estimate memory for 50M vectors at 768 dimensions in float32, then with product quantization to 32 bytes per vector.
2. Design a chunking strategy for technical documentation with headings, code blocks, and tables, and justify the overlap.
3. Interview: A RAG assistant confidently answers with outdated policy text. Where in the pipeline do you look? (Hint: the model can only use what it was given — think about what governs which chunk was retrieved and whether superseded documents were ever removed.)

Monitoring Model Drift

Software fails loudly; models fail quietly. The code is unchanged, latency and error rates are flat, and predictions keep flowing — but the world the model was trained on has moved. A fraud model meets a new attack pattern, a recommender meets a seasonal shift, a pricing model meets a competitor's promotion. Nothing pages anyone. The purpose of drift monitoring is to detect degradation before the business metric does.

Distinguish the two kinds precisely, because they call for different responses. Data drift (covariate shift) means the input distribution changed: a new device type, a new market, a broken upstream feature. Concept drift means the relationship between inputs and the label changed: the same inputs now imply a different outcome. Data drift can be detected immediately from inputs alone; concept drift generally requires labels, which arrive later — sometimes much later.

 MONITORING LAYERS, FASTEST TO SLOWEST SIGNAL

  1. INPUT / FEATURE MONITORING       minutes    nulls, ranges, category mix,
                                                 PSI vs the training baseline
  2. PREDICTION MONITORING            minutes    score distribution, positive
                                                 rate, confidence histogram
  3. PROXY / ENGAGEMENT METRICS       hours      CTR, acceptance, override rate
  4. LABEL-BASED PERFORMANCE          days-weeks AUC, precision/recall once
                                                 ground truth arrives

 A drop in 4 that was not preceded by a signal in 1-3 means your monitoring is
 blind, not that the model failed suddenly.

import math

def population_stability_index(expected: list[float], actual: list[float],
                               bins: int = 10) -> float:
    """PSI compares a feature's distribution now against training.
    Rules of thumb: <0.1 stable, 0.1-0.25 moderate shift, >0.25 significant."""
    lo, hi = min(expected), max(expected)
    width = (hi - lo) / bins or 1e-9

    def hist(xs):
        counts = [0] * bins
        for x in xs:
            i = min(int((x - lo) / width), bins - 1)
            counts[max(i, 0)] += 1
        # Floor at a small epsilon: an empty bin would make the log blow up.
        return [max(c / len(xs), 1e-6) for c in counts]

    e, a = hist(expected), hist(actual)
    return sum((ai - ei) * math.log(ai / ei) for ei, ai in zip(e, a))

import random
random.seed(1)
train = [random.gauss(50, 10) for _ in range(10_000)]
same  = [random.gauss(50, 10) for _ in range(10_000)]
moved = [random.gauss(58, 12) for _ in range(10_000)]
print(round(population_stability_index(train, same), 3))    # 0.002 -> stable
print(round(population_stability_index(train, moved), 3))   # 0.35  -> investigate

Detection is only useful with a defined response, so drift monitoring should be wired to actions rather than to a dashboard. A feature whose PSI crosses a threshold is often an upstream data bug, not a real-world change — check the pipeline before blaming the world. Sustained drift with declining proxy metrics triggers retraining on recent data. Severe drift, or an input distribution the model never saw, should trigger a fallback: route to a simpler, more robust model or to a rules-based default rather than serving confident nonsense.

Labels deserve their own design. Some arrive naturally with a delay (a chargeback in 30 days, a subscription renewal in a year); some require deliberate collection (human review of a sample); and some are contaminated by the model itself — if the fraud model blocks a transaction, you never learn whether it was fraudulent, which biases every future training set. Reserving a small unaffected holdout, or logging counterfactual slices, is what keeps the label stream honest.

Finally, monitor segments, not just aggregates. Overall AUC can hold steady while performance collapses for a new market, a new device, or a minority class — aggregate metrics hide exactly the failures that matter most for fairness and for growth areas. Set up per-segment tracking for the dimensions you care about before you need it.

Key Takeaways

• Models degrade silently: no errors, no latency change, just worse decisions.
• Data drift shifts inputs and is detectable immediately; concept drift shifts the input-label relationship and usually waits on labels.
• Monitor in layers — inputs, predictions, proxy metrics, then label-based performance — because each is progressively slower.
• PSI and similar distribution tests give a practical, thresholdable signal; check for upstream data bugs before concluding the world changed.
• Define the response in advance: investigate, retrain, or fall back to a simpler model.
• Track per-segment performance; aggregates hide localized collapse.
• Beware label feedback loops — a model that blocks outcomes stops observing them.

🧪 Practice

1. Compute PSI for a feature before and after a deliberate distribution shift, and choose an alert threshold you would defend.
2. Design the monitoring layers for a credit risk model whose labels arrive 90 days late, naming what you watch in the meantime.
3. Interview: A fraud model's precision looks stable but complaints are rising. What could be happening? (Hint: think about which cases you still get labels for once the model starts blocking, and what that does to the metric you are measuring.)

---

<a id="15-applied-system-design-case-studies"></a>