5. Data Acquisition and Storage

Every analysis begins with getting data out of somewhere else and putting it somewhere you can work with it, and the choices made at that boundary determine what questions remain answerable later. This chapter covers where data comes from, the systems that hold it, and the pipelines that move it, with attention to the failure modes that quietly corrupt results rather than producing errors.

<a id="51-data-sources"></a>

5.1 Data Sources

Data arrives in a handful of recognisable shapes, each with its own encoding hazards, access pattern, and set of things it silently discards; this section covers those shapes and what to check before trusting any of them.

Flat Files: CSV, JSON, XML, Parquet

The file is the simplest possible data interface: no server, no protocol, no credentials. That simplicity is why flat files persist despite every format's limitations, and why knowing those limitations matters.

CSV is text with delimiters. Its appeal is universality and its problem is that it carries no type information. Every value is a string until something guesses otherwise, and the guessing is where data gets damaged: a product code of 00123 loses its leading zeros, a date is read as day-month in one country and month-day in another, and a large identifier becomes a float in scientific notation. CSV also has no standard. Quoting, escaping, line endings, and encoding all vary, which is why a file that opens cleanly in one tool fails in another.

JSON adds types and nesting. It distinguishes numbers, strings, booleans, and null, and it represents hierarchy naturally, which is why it dominates web APIs. The costs are size, since keys repeat on every record, and the absence of a date type, so timestamps travel as strings by convention.

XML predates JSON and remains entrenched in finance, healthcare, and government. Its advantage is genuine schema validation through XSD plus attributes and namespaces. Its cost is verbosity and parsing complexity.

Parquet is the one designed for analytics. It is binary, columnar, typed, and compressed, it stores the schema inside the file, and it keeps per-column statistics that let a reader skip blocks that cannot match a filter. For any data you will read more than once, it is the correct default.

The analogy: CSV is a handwritten note, readable by anyone and ambiguous about everything. JSON is a filled-in form, structured but wordy. Parquet is a filing cabinet with a typed index, opaque to the eye and immediate to query.

import pandas as pd
import numpy as np
import json, io, os

# ---- CSV: the type problem, demonstrated ----
raw = """product_id,quantity,price,launch_date,notes
00123,5,19.99,03/04/2024,"Contains a comma, and ""quotes"""
00456,,24.50,04/03/2024,plain
"""
naive = pd.read_csv(io.StringIO(raw))
print(naive.dtypes.to_dict())
print(naive["product_id"].tolist())      # [123, 456]: leading zeros GONE
print(naive["quantity"].tolist())        # [5.0, nan]: int became float

# The fix is to declare everything rather than let inference guess.
typed = pd.read_csv(
    io.StringIO(raw),
    dtype={"product_id": "string", "quantity": "Int64", "price": "float64"},
    parse_dates=["launch_date"],
    date_format="%d/%m/%Y",              # NEVER rely on inference here
    keep_default_na=True,
    na_values=["", "NA", "N/A", "null", "-", "NULL"],
)
print(typed.dtypes.to_dict())
print(typed["product_id"].tolist())      # ['00123', '00456']: preserved

# THE AMBIGUOUS DATE: 03/04/2024 is 3 April or 4 March, and nothing in
# the file says which. If the first 1000 rows happen to have day <= 12,
# inference picks one convention and silently applies it to all rows.
print(pd.to_datetime("03/04/2024", dayfirst=True))    # 2024-04-03
print(pd.to_datetime("03/04/2024", dayfirst=False))   # 2024-03-04

# ENCODING: the second most common CSV failure.
data = "name,city\nJosé,Köln\n".encode("utf-8")
print(pd.read_csv(io.BytesIO(data), encoding="utf-8")["name"][0])
try:
    pd.read_csv(io.BytesIO(data), encoding="ascii")
except UnicodeDecodeError as e:
    print("decode failed:", str(e)[:48])
# Detect rather than guess when the source is unknown:
#   import charset_normalizer; charset_normalizer.from_path(p).best().encoding

# ---- JSON: typed and nested ----
payload = """
{"orders": [
  {"id": 1, "customer": {"name": "Alice", "tier": "gold"},
   "items": [{"sku": "A1", "qty": 2}, {"sku": "B2", "qty": 1}],
   "total": 59.90, "shipped": true, "coupon": null},
  {"id": 2, "customer": {"name": "Bob", "tier": "standard"},
   "items": [{"sku": "A1", "qty": 5}], "total": 99.50, "shipped": false}
]}
"""
doc = json.loads(payload)

# json_normalize flattens nested objects into dotted column names.
flat = pd.json_normalize(doc["orders"], sep=".")
print(flat[["id", "customer.name", "customer.tier", "total"]])

# record_path explodes a nested LIST into one row per element.
items = pd.json_normalize(
    doc["orders"],
    record_path="items",                 # the list to explode
    meta=["id", ["customer", "name"]],   # parent fields to carry down
)
print(items)

# JSON LINES: one object per line, so a huge file streams.
jsonl = "\n".join(json.dumps(o) for o in doc["orders"])
print(pd.read_json(io.StringIO(jsonl), lines=True).shape)
# Plain JSON requires parsing the whole document before any record is
# available. JSONL is the format to request for anything large.

# ---- Parquet: typed, columnar, compressed ----
df = pd.DataFrame({
    "product_id": pd.array(["00123", "00456"] * 5000, dtype="string"),
    "quantity": pd.array([5, None] * 5000, dtype="Int64"),
    "price": np.random.default_rng(0).random(10000) * 100,
    "launch_date": pd.to_datetime("2024-01-01") + pd.to_timedelta(
        np.arange(10000) % 365, "D"),
})

csv_path, pq_path = "/tmp/claude-0/t.csv", "/tmp/claude-0/t.parquet"
df.to_csv(csv_path, index=False)
df.to_parquet(pq_path, index=False, compression="snappy")
print(f"csv     {os.path.getsize(csv_path)/1e6:6.3f} MB")
print(f"parquet {os.path.getsize(pq_path)/1e6:6.3f} MB")

# Types survive the round trip, which CSV cannot promise.
back = pd.read_parquet(pq_path)
print(back.dtypes.to_dict())
print(back["product_id"][0], back["quantity"].isna().sum())

# Columnar means reading two columns costs two columns of IO.
print(pd.read_parquet(pq_path, columns=["price"]).shape)

# PREDICATE PUSHDOWN: the filter is applied while reading, using the
# per-row-group statistics stored in the file footer.
recent = pd.read_parquet(
    pq_path, filters=[("launch_date", ">=", pd.Timestamp("2024-06-01"))]
)
print(f"{len(recent)} of {len(df)} rows read from disk")

# INSPECT A PARQUET FILE WITHOUT READING IT.
import pyarrow.parquet as pq_mod
meta = pq_mod.read_metadata(pq_path)
print(f"{meta.num_rows} rows, {meta.num_row_groups} row groups")
print(pq_mod.read_schema(pq_path))

  CHOOSING A FLAT FILE FORMAT

                  CSV        JSON       XML        Parquet
  human readable  yes        yes        partly     no
  types stored    NO         partial    via XSD    yes, fully
  nesting         no         yes        yes        yes (structs/lists)
  schema in file  no         no         separate   yes
  compression     external   external   external   built in, per column
  column subset   no         no         no         yes
  filter on read  no         no         no         yes (statistics)
  append          trivial    awkward    awkward    new file/partition

  RULE OF THUMB
    receiving from a human or a legacy system   -> CSV, and pin the dtypes
    receiving from a web API                    -> JSON, request JSONL if large
    exchanging with finance/health/government   -> XML, validate against the XSD
    anything you will read more than once       -> convert to Parquet NOW

Key Takeaways

• CSV stores no types, so declare dtypes, date formats, and null tokens explicitly rather than letting inference guess.
• Ambiguous date orders and lost leading zeros are silent CSV corruptions that no error message reports.
• JSON carries types and nesting but has no date type; request JSON Lines for anything that must stream.
• Parquet stores schema, types, compression, and per-column statistics in the file, enabling column pruning and filter pushdown.
• Convert any repeatedly read source to Parquet once, at ingestion.

🧪 Practice

1. Write a CSV containing a zero-padded identifier, an ambiguous date, and an empty numeric cell, then read it twice, once naively and once with full dtype declarations, and list every difference.
2. Take a nested JSON document with a list field and produce both a one-row-per-record table and a one-row-per-list-element table.
3. Interview: A partner sends a 2 GB CSV daily and your load takes 40 minutes. What would you change first, and what does it cost you? (Hint: what has to happen to every byte of a text file before any column is usable?)

Relational Databases

Most operational data lives in a relational database, and reading from one is different from reading a file in ways that matter for both correctness and speed.

The essential shift in thinking is that the database is a computer, not a filing cabinet. It can filter, join, aggregate, and sort far faster than your Python process can, on data it already has locally indexed. Pulling a hundred million rows across the network to aggregate them in pandas is the most common performance mistake in analytical work. Push the work down.

Access goes through two layers. A driver speaks the database's wire protocol: psycopg for PostgreSQL, pymysql for MySQL. SQLAlchemy sits above drivers and gives one interface across engines plus connection pooling. Pandas accepts either, and for anything beyond a one-off script the SQLAlchemy engine is the right handle.

Three practices separate working code from safe code. Parameterize queries rather than formatting strings, which prevents SQL injection and lets the database cache the plan. Keep credentials in the environment, never in source. And chunk large reads, because a result set that does not fit in memory will kill the process rather than degrade gracefully.

The last subtlety is transaction isolation. A long-running analytical query sees a consistent snapshot in PostgreSQL, but in other engines or isolation levels it may see rows committed midway, producing totals that never existed at any single moment.

import os
import pandas as pd
from sqlalchemy import create_engine, text

# CREDENTIALS FROM THE ENVIRONMENT. Never in source, never in a notebook.
url = (
    f"postgresql+psycopg://{os.environ['DB_USER']}:{os.environ['DB_PASSWORD']}"
    f"@{os.environ['DB_HOST']}:5432/{os.environ.get('DB_NAME', 'analytics')}"
)
engine = create_engine(
    url,
    pool_size=5,              # reuse connections; opening one is expensive
    pool_pre_ping=True,       # detect connections the server already closed
    connect_args={"options": "-c statement_timeout=300000"},   # 5 min cap
)
# A statement timeout is the cheapest protection against a runaway query
# holding locks or saturating the warehouse for everyone else.

# ---- PUSH WORK DOWN: the single most important habit ----

# WRONG: transfer everything, then aggregate locally.
# df = pd.read_sql("SELECT * FROM orders", engine)          # 100M rows
# result = df.groupby("region")["amount"].sum()

# RIGHT: the database aggregates, you receive three rows.
result = pd.read_sql(
    text("""
        SELECT region,
               SUM(amount)                AS total,
               COUNT(*)                   AS n_orders,
               COUNT(DISTINCT customer_id) AS n_customers
        FROM orders
        WHERE order_date >= :start AND status = 'completed'
        GROUP BY region
    """),
    engine,
    params={"start": "2024-01-01"},       # PARAMETERIZED, not f-string
)

# WHY PARAMETERS, not string formatting:
user_input = "2024-01-01'; DROP TABLE orders; --"
# f"... WHERE d >= '{user_input}'"   -> executes the injected statement.
# Parameters are sent separately from the SQL text, so the value can
# never be parsed as code. This also lets the server reuse the plan.

# ---- CHUNKED READS for result sets larger than memory ----
total, rows = 0.0, 0
with engine.connect().execution_options(stream_results=True) as conn:
    for chunk in pd.read_sql(
        text("SELECT customer_id, amount FROM orders WHERE order_date >= :d"),
        conn, params={"d": "2024-01-01"}, chunksize=100_000,
    ):
        total += chunk["amount"].sum()    # accumulate a decomposable stat
        rows += len(chunk)
print(f"{rows:,} rows, total {total:,.2f}")
# stream_results=True uses a server-side cursor. Without it, the driver
# buffers the ENTIRE result client-side before pandas sees the first row,
# and chunksize only controls how it is handed to you afterwards.

# ---- INSPECTING THE SCHEMA before writing any query ----
from sqlalchemy import inspect
insp = inspect(engine)
print(insp.get_table_names(schema="public")[:10])
for col in insp.get_columns("orders"):
    print(f"  {col['name']:<20} {str(col['type']):<18} null={col['nullable']}")
print("pk:", insp.get_pk_constraint("orders")["constrained_columns"])
print("fks:", [(f["constrained_columns"], f["referred_table"])
                for f in insp.get_foreign_keys("orders")])
print("indexes:", [i["column_names"] for i in insp.get_indexes("orders")])
# Knowing which columns are indexed tells you which filters will be fast.

# ---- WRITING BACK ----
summary = result.copy()
summary.to_sql(
    "region_summary", engine, schema="analytics",
    if_exists="replace",      # 'append' | 'replace' | 'fail'
    index=False,
    chunksize=10_000,
    method="multi",           # multi-row INSERT: far fewer round trips
)
# For large loads, the row-by-row path is unusably slow. Prefer the
# database's bulk loader: COPY in PostgreSQL, LOAD DATA in MySQL.

def copy_from_dataframe(df, table, engine):
    """PostgreSQL COPY: typically 10-50x faster than to_sql."""
    import io
    buf = io.StringIO()
    df.to_csv(buf, index=False, header=False)
    buf.seek(0)
    raw = engine.raw_connection()
    try:
        with raw.cursor() as cur:
            cur.copy_expert(f"COPY {table} FROM STDIN WITH CSV", buf)
        raw.commit()
    finally:
        raw.close()

# ---- TRANSACTIONS: all or nothing ----
with engine.begin() as conn:              # commits on success, rolls back on error
    conn.execute(text("DELETE FROM region_summary WHERE run_date = :d"),
                 {"d": "2024-06-01"})
    conn.execute(text("INSERT INTO region_summary SELECT * FROM staging"))
# If the INSERT fails, the DELETE is undone. Without a transaction you
# would be left with the old rows deleted and no new rows written.

# ---- TYPE MAPPING SURPRISES ----
# NUMERIC/DECIMAL -> Python Decimal objects, so the column is dtype
#   object and arithmetic is slow. Cast in SQL: amount::float8.
# TIMESTAMPTZ     -> tz-aware; TIMESTAMP -> naive. Mixing them in a
#   comparison silently shifts times.
# NULL in an integer column -> float64 in pandas unless dtype_backend
#   is set. Use dtype_backend="pyarrow" to keep nullable integers.
recent = pd.read_sql(text("SELECT * FROM orders LIMIT 5"), engine,
                     dtype_backend="pyarrow")

Key Takeaways

• Treat the database as a compute engine and push filtering, joining, and aggregation down rather than transferring rows to pandas.
• Parameterize every query, which prevents injection and lets the server reuse execution plans.
• Use stream_results=True with chunksize, or the driver buffers the whole result set before chunking has any effect.
• Inspect the schema and indexes before writing a query, since indexed columns determine which filters are cheap.
• Wrap multi-statement writes in a transaction, and use the database's bulk loader instead of row-by-row inserts.

🧪 Practice

1. Write the same aggregation twice, once fetching all rows and computing in pandas and once in SQL, then compare elapsed time and peak memory.
2. Write a chunked reader with a server-side cursor that computes a per-group total over a table larger than your available memory.
3. Interview: A colleague builds a query with an f-string containing a user-supplied date. Name two problems beyond security. (Hint: think about what the database caches, and what happens to a quote character.)

REST APIs and Authentication

An API is a data source that answers questions rather than handing over files. That framing explains both its advantages, since you get fresh data filtered to what you asked for, and its constraints, since you get it on the provider's terms.

The dominant style is REST over HTTP. A resource has a URL, a verb says what to do with it, and the response carries a status code plus a body, usually JSON. For reading data you almost always need only GET, and the status code matters more than people expect: 200 succeeded, 429 means you are being rate limited, 4xx means your request was wrong and retrying will not help, 5xx means the server failed and retrying might.

Authentication comes in four common shapes. An API key in a header is the simplest. HTTP Basic sends a username and password. A bearer token carries a credential issued by a login step. OAuth 2 exchanges a client id and secret for a short-lived access token, which is what most large providers require. The pattern to internalise is that tokens expire, so production code must refresh rather than assume.

Pagination is the constraint that shapes your code. No API returns a million records at once. Offset pagination is simple and breaks when rows are inserted mid-walk. Cursor pagination hands you an opaque pointer to continue from and is stable. Always read the pagination section of the docs before writing the loop.

Rate limits must be handled rather than avoided. The correct response to 429 is to read the Retry-After header and wait, and the correct general retry strategy is exponential backoff with jitter, which prevents many clients from synchronising into a thundering herd.

import os, time, random
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import pandas as pd

# ---- A SESSION with sane defaults, reused across every call ----
def make_session(token: str) -> requests.Session:
    s = requests.Session()
    s.headers.update({
        "Authorization": f"Bearer {token}",
        "Accept": "application/json",
        "User-Agent": "analytics-pipeline/1.0 (data@example.com)",
    })
    retry = Retry(
        total=5,
        backoff_factor=1.0,                     # 1s, 2s, 4s, 8s, 16s
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"],                # never auto-retry writes
        respect_retry_after_header=True,
    )
    s.mount("https://", HTTPAdapter(max_retries=retry, pool_maxsize=10))
    return s
# A Session reuses the TCP connection, which on a few hundred calls is
# the difference between seconds and minutes.

session = make_session(os.environ["API_TOKEN"])

# ---- STATUS CODES decide whether retrying is even sensible ----
def fetch(url, **params):
    r = session.get(url, params=params, timeout=(5, 30))   # connect, read
    if r.status_code == 429:
        wait = int(r.headers.get("Retry-After", 60))
        time.sleep(wait)
        return fetch(url, **params)
    if 400 <= r.status_code < 500:
        # YOUR request is wrong: bad params, expired token, no permission.
        # Retrying identical input will fail identically.
        raise ValueError(f"{r.status_code}: {r.text[:200]}")
    r.raise_for_status()                          # 5xx -> raise, retry above
    return r.json()

# ALWAYS SET A TIMEOUT. Without one, requests waits forever and a
# scheduled job hangs silently instead of failing.

# ---- PAGINATION: the two patterns ----
def paginate_offset(url, page_size=100, max_pages=1000):
    """Offset paging. Simple, and unstable if rows are inserted mid-walk."""
    for page in range(max_pages):
        batch = fetch(url, limit=page_size, offset=page * page_size)
        records = batch.get("data", [])
        if not records:
            return
        yield from records
        if len(records) < page_size:              # short page = last page
            return

def paginate_cursor(url, page_size=100):
    """Cursor paging. Stable under concurrent writes; prefer it."""
    cursor = None
    while True:
        params = {"limit": page_size}
        if cursor:
            params["cursor"] = cursor
        batch = fetch(url, **params)
        yield from batch.get("data", [])
        cursor = batch.get("next_cursor")
        if not cursor:
            return

# Some APIs use RFC 5988 Link headers instead of a body field.
def paginate_link_header(url):
    while url:
        r = session.get(url, timeout=(5, 30))
        r.raise_for_status()
        yield from r.json()
        url = r.links.get("next", {}).get("url")

# ---- OAUTH 2 CLIENT CREDENTIALS, with refresh ----
class OAuthClient:
    def __init__(self, token_url, client_id, client_secret):
        self.token_url, self.cid, self.secret = token_url, client_id, client_secret
        self._token, self._expires_at = None, 0.0

    def token(self) -> str:
        if self._token and time.time() < self._expires_at - 60:   # 60s margin
            return self._token
        r = requests.post(self.token_url, data={
            "grant_type": "client_credentials",
            "client_id": self.cid, "client_secret": self.secret,
        }, timeout=10)
        r.raise_for_status()
        payload = r.json()
        self._token = payload["access_token"]
        self._expires_at = time.time() + payload.get("expires_in", 3600)
        return self._token
# The 60-second margin matters: a token that expires between your check
# and the server's check produces an intermittent 401 that is painful to
# reproduce.

# ---- EXPONENTIAL BACKOFF WITH JITTER, written out ----
def backoff_get(url, attempts=5, base=1.0, cap=60.0):
    for i in range(attempts):
        try:
            r = session.get(url, timeout=(5, 30))
            if r.status_code < 500 and r.status_code != 429:
                r.raise_for_status()
                return r.json()
        except requests.RequestException:
            if i == attempts - 1:
                raise
        delay = min(cap, base * 2 ** i) * (0.5 + random.random())   # jitter
        time.sleep(delay)
    raise RuntimeError("exhausted retries")
# Without jitter, every client that failed at the same moment retries at
# the same moment, re-creating the overload that caused the failure.

# ---- RESPECTING A RATE LIMIT PROACTIVELY ----
class RateLimiter:
    def __init__(self, calls_per_second):
        self.interval = 1.0 / calls_per_second
        self.last = 0.0
    def wait(self):
        gap = time.time() - self.last
        if gap < self.interval:
            time.sleep(self.interval - gap)
        self.last = time.time()

# ---- PERSIST THE RAW RESPONSE before parsing ----
records = list(paginate_cursor("https://api.example.com/v1/orders"))
raw_df = pd.DataFrame(records)
raw_df.to_parquet("data/raw/orders_2024-06-01.parquet")
# Keeping the untransformed payload means a parsing bug found next month
# can be fixed WITHOUT re-calling an API that may no longer serve
# historical data, may rate-limit you, or may have changed its schema.

flat = pd.json_normalize(records, sep=".")
print(flat.columns.tolist()[:8])

  WHAT THE STATUS CODE TELLS YOU TO DO

  2xx  success                       parse the body
  301/302 redirect                   follow (requests does by default)
  400  bad request                   fix your parameters; do NOT retry
  401  unauthenticated               refresh the token, then retry once
  403  authenticated, not allowed    a permissions problem; do NOT retry
  404  not found                     the resource does not exist
  422  validation failed             your payload is malformed
  429  rate limited                  read Retry-After and WAIT
  5xx  server error                  retry with exponential backoff

  THE RULE: 4xx means change something. 5xx and 429 mean wait.

  PAGINATION

  OFFSET                          CURSOR
  ?limit=100&offset=0             ?limit=100
  ?limit=100&offset=100           ?limit=100&cursor=eyJpZCI6MTAwfQ
  ?limit=100&offset=200           ?limit=100&cursor=eyJpZCI6MjAwfQ

  simple, any page reachable      opaque, sequential only
  a row inserted at position 50   stable: the cursor points at a record,
  shifts everything, so you       so inserts do not shift your position
  re-read one row and MISS one

Key Takeaways

• Status codes dictate the response: 4xx means fix the request, while 429 and 5xx mean wait and retry.
• Always set an explicit timeout, or a scheduled job hangs instead of failing visibly.
• Prefer cursor pagination over offset, because inserts during a walk shift offsets and cause skipped or duplicated rows.
• Refresh OAuth tokens before they expire, with a safety margin, to avoid intermittent and hard-to-reproduce 401 errors.
• Persist the raw response before parsing, so a later parsing fix does not require re-fetching data the API may no longer serve.

🧪 Practice

1. Write a paginated fetcher that handles both a next_cursor body field and a Link header, yielding records lazily.
2. Implement exponential backoff with jitter and demonstrate, with a simulated failing endpoint, that the delays grow and vary.
3. Interview: Your nightly API job succeeds most nights and fails with 401 roughly once a week at varying times. What is your first hypothesis? (Hint: what has a fixed lifetime and is checked twice?)

Web Scraping Fundamentals

Scraping extracts data from pages built for human eyes rather than machine consumption. It is the acquisition method of last resort, because it is fragile, frequently restricted, and sometimes unnecessary.

Before writing any scraper, check three things in order. Is there an API? Is there a bulk download or dataset export? Is there a hidden JSON endpoint that the page itself calls? That last one is the most commonly missed: open the browser network tab, filter to XHR, and you will often find the page fetching clean JSON that you can request directly, which is faster, more stable, and easier to parse than the rendered HTML.

Legality and ethics are genuine constraints, not formalities. Respect robots.txt, read the site's terms of service, identify yourself in the User-Agent, rate-limit yourself well below what the server can handle, and never scrape personal data without a lawful basis. Scraping public factual data for analysis is generally defensible; circumventing access controls or republishing copyrighted content is not.

Technically there are two cases. Static HTML arrives complete and requests plus a parser suffices. JavaScript-rendered pages arrive nearly empty and populate in the browser, requiring a headless browser such as Playwright, which is far slower. Check by comparing the page source against what you see rendered.

The parsing discipline that matters is selector stability. A class name like css-1x7fj2k is generated by a build tool and will change next deployment. Anchor on semantic attributes, id, or text content instead, and make every extraction tolerate absence rather than raising.

import time, random, re
from urllib.parse import urljoin, urlparse
from urllib.robotparser import RobotFileParser
import requests
from bs4 import BeautifulSoup
import pandas as pd

# ---- STEP 0: CHECK WHETHER YOU SHOULD, AND WHETHER YOU NEED TO ----
def may_fetch(url, agent="analytics-bot"):
    parts = urlparse(url)
    rp = RobotFileParser()
    rp.set_url(f"{parts.scheme}://{parts.netloc}/robots.txt")
    try:
        rp.read()
    except Exception:
        return True, None                 # no robots.txt: no stated restriction
    return rp.can_fetch(agent, url), rp.crawl_delay(agent)

allowed, delay = may_fetch("https://example.com/products")
print(f"allowed={allowed}, crawl_delay={delay}")

# BEFORE SCRAPING HTML, look for the JSON the page itself uses:
#   browser devtools -> Network -> filter XHR -> reload
# A request to /api/products?page=1 returning clean JSON is far better
# than parsing the rendered table, and it is usually right there.

# ---- A POLITE SESSION ----
session = requests.Session()
session.headers.update({
    "User-Agent": "analytics-bot/1.0 (+https://example.org/bot; data@example.org)",
    "Accept": "text/html,application/xhtml+xml",
})
# Identify yourself honestly. An operator who can contact you will email
# before blocking you.

def polite_get(url, min_delay=1.0):
    time.sleep(min_delay + random.random())       # never hammer
    r = session.get(url, timeout=(5, 20))
    r.raise_for_status()
    return r

# ---- PARSING STATIC HTML ----
html = """
<html><body>
  <div class="product-list">
    <article class="css-1x7fj2k" data-product-id="A1">
      <h2 class="title">Widget</h2>
      <span class="price" data-amount="19.99">$19.99</span>
      <span class="stock">In stock</span>
      <a href="/products/a1">details</a>
    </article>
    <article class="css-1x7fj2k" data-product-id="B2">
      <h2 class="title">Gadget</h2>
      <span class="price" data-amount="24.50">$24.50</span>
      <a href="/products/b2">details</a>
    </article>
  </div>
</body></html>
"""
soup = BeautifulSoup(html, "lxml")               # lxml is faster than html.parser

# SELECTOR CHOICE decides how long the scraper survives.
# FRAGILE: a build-generated class changes on the next deployment.
fragile = soup.select("article.css-1x7fj2k")
# STABLE: a semantic data attribute the developers chose deliberately.
stable = soup.select("article[data-product-id]")
print(len(fragile), len(stable))

def extract(article, base="https://example.com"):
    """Every field tolerates absence. Never assume an element exists."""
    def text_of(sel):
        el = article.select_one(sel)
        return el.get_text(strip=True) if el else None

    price_el = article.select_one("[data-amount]")
    link_el = article.select_one("a[href]")
    return {
        "product_id": article.get("data-product-id"),
        "title": text_of("h2"),
        # Prefer a machine-readable attribute over parsing display text,
        # which carries currency symbols, thousands separators, and locale.
        "price": float(price_el["data-amount"]) if price_el else None,
        "stock": text_of(".stock"),               # absent in the second item
        "url": urljoin(base, link_el["href"]) if link_el else None,
    }

rows = [extract(a) for a in stable]
df = pd.DataFrame(rows)
print(df)
print(df["stock"].tolist())                       # ['In stock', None]

# PARSING DISPLAY TEXT when no attribute exists.
def parse_price(s):
    if not s:
        return None
    m = re.search(r"[\d.,]+", s.replace(",", ""))  # strip thousands separators
    return float(m.group()) if m else None
print(parse_price("$1,299.00"))

# ---- JAVASCRIPT-RENDERED PAGES ----
# Diagnose first: if the data is absent from the raw source, it is rendered.
#   "Widget" in requests.get(url).text   -> False means you need a browser.
#
# from playwright.sync_api import sync_playwright
# with sync_playwright() as p:
#     browser = p.chromium.launch(headless=True)
#     page = browser.new_page()
#     page.goto(url, wait_until="networkidle")
#     page.wait_for_selector("article[data-product-id]")   # wait for CONTENT,
#     html = page.content()                                # not a fixed sleep
#     browser.close()
#
# Playwright is roughly 50-100x slower per page than requests. Use it only
# after confirming there is no underlying JSON endpoint.

# ---- CACHE EVERY FETCH ----
from pathlib import Path
import hashlib

def cached_get(url, cache_dir="data/raw/html"):
    Path(cache_dir).mkdir(parents=True, exist_ok=True)
    key = hashlib.sha256(url.encode()).hexdigest()[:16]
    path = Path(cache_dir) / f"{key}.html"
    if path.exists():
        return path.read_text()
    text = polite_get(url).text
    path.write_text(text)
    return text
# Caching means a parser bug costs you a re-parse, not a re-crawl. It also
# keeps you off the server while you iterate, which is the polite part.

# ---- MONITOR FOR SILENT BREAKAGE ----
def scrape_with_checks(urls, expected_min=1):
    out, failures = [], []
    for u in urls:
        try:
            page = BeautifulSoup(cached_get(u), "lxml")
            items = page.select("article[data-product-id]")
            if len(items) < expected_min:
                failures.append((u, f"found {len(items)} items"))
                continue
            out.extend(extract(i) for i in items)
        except Exception as e:
            failures.append((u, str(e)[:60]))
    if failures:
        print(f"WARNING: {len(failures)} of {len(urls)} pages failed")
    return pd.DataFrame(out), failures
# A redesigned site does not raise an exception. It returns zero rows,
# and without this check the pipeline reports success on an empty table.

Key Takeaways

• Check for an API, a bulk export, or a hidden JSON endpoint before parsing HTML, because all three are more stable and faster.
• Respect robots.txt and terms of service, identify yourself in the User-Agent, and rate-limit well below capacity.
• Anchor selectors on semantic attributes rather than build-generated class names, which change on every deployment.
• Make every field extraction tolerate a missing element, and prefer machine-readable attributes over displayed text.
• Cache raw HTML so parser iteration costs no further requests, and assert a minimum row count because a redesign returns zero rows, not an error.

🧪 Practice

1. Write a scraper for a static page that extracts four fields, returning None for any missing element rather than raising.
2. Add disk caching and a minimum-row assertion, then simulate a layout change and confirm the check fires.
3. Interview: A scraper that ran nightly for six months suddenly writes an empty table with no error. What happened and what would have caught it? (Hint: what does a CSS selector return when it matches nothing?)

Streaming and Event Sources

Batch data answers "what happened yesterday". Streaming data answers "what is happening now", and the difference is architectural rather than a matter of running batches more often.

The shift is from state to events. A database table holds the current state: this customer's balance is 340. An event stream holds the sequence of things that occurred: deposited 500, withdrew 160. State is derivable from events, but events are not recoverable from state, which is why event streams preserve information that tables discard.

The dominant model is the append-only log. Producers append records to a topic, which is split into partitions for parallelism. Consumers read at their own pace, tracking an offset. Crucially, reading does not remove the record, so many independent consumers can read the same stream and a new consumer can replay from the beginning.

Ordering is guaranteed only within a partition. Records land in a partition by key, so all events for one customer stay ordered if you key by customer id. Choose that key deliberately: key by something too coarse and one partition becomes a bottleneck; key by something too fine and ordering guarantees you needed are lost.

The concept that causes the most confusion is the difference between event time and processing time. An event occurs at 10:00 and arrives at 10:07 because a phone was offline. Aggregating by arrival time puts it in the wrong bucket. Correct stream processing uses event time with a watermark, a declaration of how late data may arrive before a window closes.

Delivery semantics are the other fundamental. At-most-once may lose records, at-least-once may duplicate them, and exactly-once requires coordination. Most systems give at-least-once, so consumers must be idempotent.

import json, time
from datetime import datetime, timezone, timedelta
from collections import defaultdict

# ---- PRODUCING to Kafka ----
# from kafka import KafkaProducer
# producer = KafkaProducer(
#     bootstrap_servers=["broker:9092"],
#     value_serializer=lambda v: json.dumps(v).encode(),
#     key_serializer=lambda k: k.encode(),
#     acks="all",            # wait for all replicas: durability over latency
#     retries=5,
#     enable_idempotence=True,   # prevents duplicates from producer retries
#     compression_type="snappy",
# )
event = {
    "event_id": "evt_01H8X",          # a unique id: the basis of dedup
    "event_type": "order_placed",
    "occurred_at": "2024-06-01T10:00:00Z",   # EVENT time, set by the producer
    "customer_id": "c_123",
    "payload": {"order_id": "o_987", "amount": 59.90},
}
# producer.send("orders", key=event["customer_id"], value=event)
# The KEY determines the partition, so keying by customer_id guarantees
# that one customer's events are processed in order.

# ---- CONSUMING, with manual offset commits ----
# consumer = KafkaConsumer(
#     "orders",
#     bootstrap_servers=["broker:9092"],
#     group_id="analytics-loader",     # the unit that shares partitions
#     auto_offset_reset="earliest",    # or "latest" to skip history
#     enable_auto_commit=False,        # commit AFTER processing, not before
#     value_deserializer=lambda b: json.loads(b.decode()),
# )
#
# for batch in iter(lambda: consumer.poll(timeout_ms=1000, max_records=500), None):
#     for tp, messages in batch.items():
#         rows = [m.value for m in messages]
#         write_to_warehouse(rows)     # must be IDEMPOTENT
#         consumer.commit()            # only after the write succeeded
#
# Committing BEFORE processing gives at-most-once: a crash loses the batch.
# Committing AFTER gives at-least-once: a crash reprocesses it. The second
# is almost always correct, and it is why the write must be idempotent.

# ---- IDEMPOTENT CONSUMPTION via a deduplication key ----
def upsert_events(rows, conn):
    """Re-delivering the same event must not change the result."""
    # INSERT INTO events (event_id, ...) VALUES (...)
    # ON CONFLICT (event_id) DO NOTHING;
    #
    # The unique constraint on event_id makes replay harmless, which
    # converts at-least-once delivery into effectively-once processing.
    ...

# ---- EVENT TIME VS PROCESSING TIME, demonstrated ----
now = datetime(2024, 6, 1, 10, 7, tzinfo=timezone.utc)
stream = [
    {"id": 1, "occurred_at": "2024-06-01T10:01:00Z", "arrived_at": "10:01"},
    {"id": 2, "occurred_at": "2024-06-01T10:03:00Z", "arrived_at": "10:03"},
    {"id": 3, "occurred_at": "2024-06-01T09:58:00Z", "arrived_at": "10:06"},  # LATE
    {"id": 4, "occurred_at": "2024-06-01T10:05:00Z", "arrived_at": "10:06"},
]

def bucket(ts_iso, minutes=5):
    t = datetime.fromisoformat(ts_iso.replace("Z", "+00:00"))
    return t.replace(minute=(t.minute // minutes) * minutes,
                     second=0, microsecond=0)

by_event_time = defaultdict(list)
for e in stream:
    by_event_time[bucket(e["occurred_at"])].append(e["id"])
for k in sorted(by_event_time):
    print(k.strftime("%H:%M"), by_event_time[k])
# 09:55 -> [3]   the late event lands in the window where it BELONGS
# 10:00 -> [1, 2, 4]
#
# Bucketing by arrival time would have put event 3 in the 10:05 window,
# overstating that window and understating the 09:55 one. Any metric
# computed from arrival time is wrong in a way that never self-corrects.

# ---- WATERMARKS: deciding when a window is final ----
class WindowManager:
    """Emit a window once the watermark passes its end."""
    def __init__(self, window_s=300, allowed_lateness_s=120):
        self.window_s = window_s
        self.lateness = allowed_lateness_s
        self.windows = defaultdict(list)
        self.max_event_time = None

    def add(self, event_time, value):
        w = int(event_time.timestamp()) // self.window_s * self.window_s
        self.max_event_time = (event_time if self.max_event_time is None
                               else max(self.max_event_time, event_time))
        watermark = self.max_event_time - timedelta(seconds=self.lateness)
        if datetime.fromtimestamp(w + self.window_s, timezone.utc) < watermark:
            return "dropped_too_late"        # count these; never ignore them
        self.windows[w].append(value)
        return "accepted"

    def closed(self):
        """Windows entirely behind the watermark are final."""
        wm = self.max_event_time - timedelta(seconds=self.lateness)
        return {w: v for w, v in self.windows.items()
                if datetime.fromtimestamp(w + self.window_s, timezone.utc) < wm}

wm = WindowManager()
for e in stream:
    t = datetime.fromisoformat(e["occurred_at"].replace("Z", "+00:00"))
    print(e["id"], wm.add(t, e["id"]))
# The tradeoff is explicit: a longer allowed lateness captures more
# stragglers and delays every result. There is no setting that does both.

# ---- CHANGE DATA CAPTURE as an event source ----
# Debezium reads the database write-ahead log and emits one event per row
# change, so the warehouse follows the source without polling it.
cdc_event = {
    "op": "u",                                   # c=create, u=update, d=delete
    "before": {"id": 7, "status": "pending", "amount": 50.0},
    "after":  {"id": 7, "status": "shipped", "amount": 50.0},
    "source": {"table": "orders", "lsn": 24580923, "ts_ms": 1717236000000},
}
print(f"{cdc_event['op']}: "
      f"{cdc_event['before']['status']} -> {cdc_event['after']['status']}")
# Note that a DELETE is visible here. A nightly snapshot of the table
# would simply show the row missing, with no record that it existed.

  STATE VERSUS EVENTS

  TABLE (state)                    LOG (events)
  customer  balance                deposit  500   10:00
  c_123     340                    withdraw 160   10:05

  answers "what is true now"       answers "what happened, and when"
  overwrites history               append only, replayable
  one consumer's view              any number of independent consumers

  State is derivable from events. Events are NOT derivable from state.

  EVENT TIME VERSUS PROCESSING TIME

  occurred  09:58 ----------------------> arrived 10:06   (phone offline)
  occurred  10:01 --> arrived 10:01
  occurred  10:05 --> arrived 10:06

  windowed by ARRIVAL:  09:55 [] , 10:05 [09:58 event, 10:05 event]  WRONG
  windowed by EVENT:    09:55 [09:58 event] , 10:05 [10:05 event]    RIGHT

  WATERMARK = max event time seen, minus allowed lateness.
  A window closes when the watermark passes its end.
    short lateness -> fast results, more dropped stragglers
    long lateness  -> complete results, delayed

Key Takeaways

• An append-only log preserves the sequence of events, from which state can be derived, while a table overwrites history.
• Ordering holds only within a partition, so the partition key determines both parallelism and which ordering guarantees you keep.
• Commit offsets after processing, not before, which gives at-least-once delivery and requires idempotent consumers.
• Aggregate by event time rather than arrival time, or late data lands in the wrong window and the error never self-corrects.
• A watermark trades completeness against latency explicitly; count dropped late records rather than ignoring them.

🧪 Practice

1. Simulate a stream with out-of-order events and compute five-minute counts by both event time and arrival time, quantifying the difference.
2. Implement an idempotent consumer using a unique event id, and show that replaying a batch leaves the output unchanged.
3. Interview: Your streaming dashboard shows a traffic spike at 10:06 that the batch report the next day does not show. Explain. (Hint: which timestamp does each system group by?)

Public and Open Datasets

Public data is the fastest route to a working prototype, a benchmark, or a denominator your internal data lacks. It is also the source where provenance problems are easiest to miss, because the data arrives looking finished.

The useful mental division is by publisher type. Government and statistical agencies publish authoritative, slow-moving, well-documented data. Research repositories publish datasets attached to papers, usually with a methodology document. Platform datasets such as those on Kaggle are convenient and frequently unlabelled as to origin, licence, or processing already applied. Commercial data vendors sell curated feeds with an SLA.

Before using any public dataset, answer four questions. Who collected it and how, because a convenience sample cannot support population claims. What licence applies, since some forbid commercial use or require attribution. When was it last updated, because a stale dataset presented as current is a common and embarrassing failure. And what is the unit of observation, which is astonishingly often undocumented.

The recurring analytical trap is coverage. A dataset of reported crimes is a dataset of reporting, not of crime. A dataset of app reviews covers people motivated enough to review. Treating a biased sampling frame as the population is the single most common error with public data, and no amount of modelling fixes it.

The engineering practice that pays off is pinning. Record the exact download URL, the date, a checksum of the file, and the version if one exists. Public datasets are revised without notice, and a result you cannot reproduce because the source changed is indistinguishable from a result you got wrong.

import hashlib, json
from pathlib import Path
from datetime import datetime, timezone
import requests
import pandas as pd

# ---- PIN EVERY DOWNLOAD: url, date, checksum, size ----
def fetch_pinned(url, dest_dir="data/external", expected_sha256=None):
    dest_dir = Path(dest_dir)
    dest_dir.mkdir(parents=True, exist_ok=True)
    name = url.rsplit("/", 1)[-1]
    path = dest_dir / name

    r = requests.get(url, timeout=60, stream=True)
    r.raise_for_status()
    h = hashlib.sha256()
    with open(path, "wb") as fh:
        for block in r.iter_content(1 << 20):      # 1 MB blocks
            h.update(block)
            fh.write(block)
    digest = h.hexdigest()

    if expected_sha256 and digest != expected_sha256:
        raise ValueError(
            f"checksum mismatch: the source changed.\n"
            f"  expected {expected_sha256}\n  got      {digest}"
        )

    manifest = {
        "url": url,
        "downloaded_at": datetime.now(timezone.utc).isoformat(),
        "sha256": digest,
        "bytes": path.stat().st_size,
        "etag": r.headers.get("ETag"),
        "last_modified": r.headers.get("Last-Modified"),
    }
    (dest_dir / f"{name}.manifest.json").write_text(json.dumps(manifest, indent=2))
    return path, manifest

# Committing the manifest to git makes "which version of the source
# produced this figure" an answerable question a year later.

# ---- PROFILE BEFORE TRUSTING ----
def profile(df: pd.DataFrame, name="dataset"):
    print(f"\n{name}: {len(df):,} rows x {df.shape[1]} columns")
    report = pd.DataFrame({
        "dtype": df.dtypes.astype(str),
        "nulls": df.isna().sum(),
        "null_pct": (df.isna().mean() * 100).round(1),
        "distinct": df.nunique(),
        "example": [df[c].dropna().iloc[0] if df[c].notna().any() else None
                    for c in df.columns],
    })
    print(report.to_string())

    # Sentinel values hide in the extremes of numeric columns.
    for c in df.select_dtypes("number"):
        s = df[c]
        suspicious = [v for v in (-1, -999, -9999, 0, 9999, 99999)
                      if (s == v).sum() > len(s) * 0.01]
        if suspicious:
            print(f"  WARNING {c}: frequent values {suspicious} may be sentinels")
    return report

sample = pd.DataFrame({
    "region": ["eu", "us", "eu", None],
    "population": [447_000_000, 331_000_000, -999, 126_000_000],
    "year": [2023, 2023, 2023, 2023],
})
profile(sample, "population_by_region")
# -999 is a documented "not reported" code in many statistical releases,
# and it will quietly become part of a mean if you do not look.

# ---- CHECK COVERAGE AGAINST A KNOWN UNIVERSE ----
def coverage_check(df, key, universe):
    present = set(df[key].dropna())
    missing = set(universe) - present
    extra = present - set(universe)
    print(f"{key}: {len(present)}/{len(universe)} covered")
    if missing:
        print(f"  MISSING: {sorted(missing)[:10]}")
    if extra:
        print(f"  UNEXPECTED: {sorted(extra)[:10]}")
    return missing, extra

coverage_check(sample, "region", {"eu", "us", "ap", "latam"})
# A missing region is not a null in any column. It is an absent ROW,
# which no null check finds and which silently biases every total.

# ---- LICENCES: check before, not after ----
LICENCES = {
    "CC0":        {"commercial": True,  "attribution": False, "share_alike": False},
    "CC-BY":      {"commercial": True,  "attribution": True,  "share_alike": False},
    "CC-BY-SA":   {"commercial": True,  "attribution": True,  "share_alike": True},
    "CC-BY-NC":   {"commercial": False, "attribution": True,  "share_alike": False},
    "ODbL":       {"commercial": True,  "attribution": True,  "share_alike": True},
    "proprietary":{"commercial": None,  "attribution": None,  "share_alike": None},
}
def usable_commercially(lic):
    terms = LICENCES.get(lic)
    if terms is None or terms["commercial"] is None:
        return "REVIEW REQUIRED"
    return "yes" if terms["commercial"] else "NO"
for lic in ["CC0", "CC-BY-NC", "ODbL", "proprietary"]:
    print(f"{lic:<12} commercial use: {usable_commercially(lic)}")
# Share-alike matters for derived work: publishing a model or a dataset
# built on ODbL data can obligate you to license the result the same way.

# ---- A DATASET CARD, written at ingestion while you still know ----
CARD = """
# Dataset: Regional Population Estimates
source_url:    https://example.gov/data/pop_2023.csv
sha256:        a3f1...  (see manifest)
retrieved:     2024-06-01
licence:       CC-BY 4.0 (attribution required in any published output)
publisher:     National Statistics Office
update_cycle:  annual, published each March for the prior year
unit:          one row per (region, year)
coverage:      4 macro regions; EXCLUDES overseas territories
known_issues:  population uses -999 for "not reported" (2 rows in 2023)
               2020 figures were revised in the 2022 release
suitable_for:  denominators in per-capita rates
NOT suitable:  sub-regional analysis, any pre-2015 trend (methodology change)
"""
Path("data/external/population.CARD.md").write_text(CARD)
# The "NOT suitable" line prevents the most expensive kind of reuse
# mistake, which is someone finding the file six months later.

Publisher type	Strengths	Watch for
Statistical agency	authoritative, documented, stable	slow release, revisions to past years
Research repository	methodology published	small samples, narrow populations
Platform (Kaggle and similar)	convenient, ready to load	unclear provenance, pre-cleaned, unclear licence
Open data portal	local and current	inconsistent schemas between releases
Commercial vendor	curated, supported, SLA	cost, redistribution restrictions

Key Takeaways

• Establish who collected the data, how, under what licence, and when, before using it for anything that will be published.
• Pin every download with a URL, timestamp, and checksum, because public datasets are revised without notice.
• Profile for sentinel values such as -999, which enter averages silently and are never flagged as nulls.
• Check coverage against a known universe, since a missing row is invisible to every null check.
• Treat the sampling frame as the real limit: a dataset of reports measures reporting, not the underlying phenomenon.

🧪 Practice

1. Write a download function that records a manifest with URL, timestamp, checksum, and size, then demonstrate it detecting a changed source.
2. Profile a public dataset and identify at least one sentinel value or coverage gap, documenting both in a short dataset card.
3. Interview: A colleague benchmarks a model on a popular public dataset and reports state-of-the-art accuracy. Name two reasons to be cautious. (Hint: consider who else has used it, and how it was prepared.)

Survey and Instrument Data

Survey and sensor data share a property that distinguishes them from transactional records: the data is generated by a measurement process you designed, so its errors are systematic and knowable rather than random.

For surveys, the governing concept is the difference between the target population, the sampling frame, and the respondents. The frame is who could be reached; respondents are who answered. Every gap between these introduces bias. Coverage error excludes people from the frame, sampling error is the random variation from taking a sample, and nonresponse error is the difference between those who answered and those who did not.

The critical consequence is that survey data almost always requires weights. If young people respond at half the rate of older people, an unweighted mean over-represents older views. Weights correct for unequal selection probability and differential response, and ignoring them produces a confidently wrong number. Weights also mean that standard errors computed by ordinary formulas are wrong, because the design is not a simple random sample.

Question design is part of the data. Leading wording, unbalanced scales, double-barrelled questions, and ordering effects all shift responses measurably. A dataset cannot be interpreted without its questionnaire.

For instrument data, the analogous concerns are calibration, drift, sampling rate, and missingness mechanism. A sensor that fails during extreme conditions produces data that is missing precisely when it matters most, which is missingness not at random and cannot be imputed away.

import numpy as np
import pandas as pd

rng = np.random.default_rng(0)

# ---- WEIGHTED ESTIMATION: why unweighted means mislead ----
# Population: 40% aged 18-34, 60% aged 35+. Younger people respond less.
pop_share = {"18-34": 0.40, "35+": 0.60}
resp = pd.DataFrame({
    "age_group": ["18-34"] * 100 + ["35+"] * 300,     # 25% / 75% of the sample
    "supports": np.r_[rng.binomial(1, 0.70, 100),     # young: 70% support
                      rng.binomial(1, 0.30, 300)],    # older: 30% support
})

sample_share = resp["age_group"].value_counts(normalize=True)
print(sample_share.round(3).to_dict())

# The design weight restores the population composition.
resp["weight"] = resp["age_group"].map(
    {g: pop_share[g] / sample_share[g] for g in pop_share}
)
print(resp.groupby("age_group")["weight"].first().round(3).to_dict())

unweighted = resp["supports"].mean()
weighted = np.average(resp["supports"], weights=resp["weight"])
truth = sum(pop_share[g] * p for g, p in [("18-34", 0.70), ("35+", 0.30)])
print(f"unweighted {unweighted:.3f} | weighted {weighted:.3f} | truth {truth:.3f}")
# The unweighted estimate is off by roughly 10 percentage points, and
# nothing in the data announces it. Only the frame information reveals it.

# ---- STANDARD ERRORS UNDER A COMPLEX DESIGN ----
def design_effect(weights):
    """Kish's approximation: how much the weighting inflates variance."""
    w = np.asarray(weights, dtype=float)
    return len(w) * (w ** 2).sum() / w.sum() ** 2

deff = design_effect(resp["weight"])
n_eff = len(resp) / deff
naive_se = np.sqrt(weighted * (1 - weighted) / len(resp))
adjusted_se = np.sqrt(weighted * (1 - weighted) / n_eff)
print(f"deff {deff:.2f}, effective n {n_eff:.0f} of {len(resp)}")
print(f"SE naive {naive_se:.4f} -> adjusted {adjusted_se:.4f}")
print(f"95% CI: [{weighted - 1.96*adjusted_se:.3f}, "
      f"{weighted + 1.96*adjusted_se:.3f}]")
# Reporting the naive interval overstates precision. A weighted survey of
# 400 may carry the precision of a simple random sample of 300.

# ---- LIKERT SCALES: ordered, not numeric ----
scale = ["strongly disagree", "disagree", "neutral", "agree", "strongly agree"]
responses = pd.Series(
    rng.choice(scale, 500, p=[0.1, 0.15, 0.2, 0.35, 0.2]), name="q1"
).astype(pd.CategoricalDtype(scale, ordered=True))

print(responses.value_counts().reindex(scale).to_dict())
print(f"top-2-box: {responses.isin(scale[3:]).mean():.1%}")     # the standard KPI
print(f"median: {responses.sort_values().iloc[len(responses)//2]}")

numeric = responses.cat.codes + 1
print(f"mean as 1-5: {numeric.mean():.2f}")
# Averaging an ordinal scale assumes the gap from "disagree" to "neutral"
# equals the gap from "agree" to "strongly agree". It does not. Report
# top-2-box or the full distribution; use the mean only for tracking
# movement over time, and say that is what you are doing.

# ---- NONRESPONSE: item versus unit ----
survey = pd.DataFrame({
    "respondent_id": range(1, 301),
    "age": rng.integers(18, 80, 300).astype(float),
    "income": rng.normal(50_000, 20_000, 300),
    "satisfaction": rng.integers(1, 6, 300).astype(float),
})
# High earners decline the income question at a higher rate: MNAR.
refuse = rng.random(300) < (survey["income"] > 65_000) * 0.5
survey.loc[refuse, "income"] = np.nan

print(f"item nonresponse on income: {survey['income'].isna().mean():.1%}")
print(f"mean of reported income: {survey['income'].mean():,.0f}")
print(f"mean of ALL income:      {50_000:,.0f} (by construction)")
# Mean imputation would make this worse, because it imputes the mean of
# the LOW earners who answered. This is missing-not-at-random, and no
# imputation strategy recovers it from this data alone.

# ---- OPEN-ENDED TEXT: keep the raw response ----
verbatim = pd.Series(["too expensive", "Too Expensive!", "price is high",
                      "great product", "  too expensive "])
normalized = (verbatim.str.strip().str.lower()
                      .str.replace(r"[^\w\s]", "", regex=True))
print(normalized.value_counts().to_dict())
# Normalization unifies format, not meaning: "price is high" still needs
# coding. Always retain the verbatim column alongside any coded one.

# ---- INSTRUMENT DATA: drift, calibration, and informative gaps ----
hours = pd.date_range("2024-06-01", periods=720, freq="h")
true_temp = 20 + 5 * np.sin(np.arange(720) * 2 * np.pi / 24)
drift = np.linspace(0, 2.5, 720)                 # the sensor slowly biases high
reading = true_temp + drift + rng.normal(0, 0.3, 720)

sensor = pd.DataFrame({"ts": hours, "reading": reading}).set_index("ts")

# The sensor cuts out above 28 degrees: missing exactly at the extremes.
sensor.loc[sensor["reading"] > 28, "reading"] = np.nan
print(f"missing: {sensor['reading'].isna().mean():.1%}")
print(f"mean of observed: {sensor['reading'].mean():.2f}")
print(f"mean of truth:    {true_temp.mean() + drift.mean():.2f}")
# Dropping the gaps understates the mean and erases every peak. For a
# heat-warning system, the missing values ARE the event of interest.

# DETECTING DRIFT: compare early and late periods against a reference.
early = sensor["reading"].iloc[:168].mean()      # first week
late = sensor["reading"].iloc[-168:].mean()      # last week
print(f"week 1 {early:.2f} -> week 4 {late:.2f} (drift suspected)")

# RANGE AND RATE CHECKS catch faults that look like data.
def validate_sensor(s, lo, hi, max_rate):
    flags = pd.DataFrame(index=s.index)
    flags["out_of_range"] = (s < lo) | (s > hi)
    flags["impossible_jump"] = s.diff().abs() > max_rate
    flags["stuck"] = s.diff().eq(0).rolling(6).sum() >= 6   # frozen reading
    return flags

flags = validate_sensor(sensor["reading"], lo=-40, hi=60, max_rate=5.0)
print(flags.sum().to_dict())
# A stuck sensor reports a perfectly plausible constant. Only the
# zero-variance check finds it.

Key Takeaways

• The gaps between target population, sampling frame, and respondents are where survey bias enters, and none of them is visible in the data file.
• Weighted estimates are usually required, and ignoring weights produces a confidently wrong number with no warning.
• A complex design inflates variance, so naive standard errors overstate precision; adjust using the design effect.
• Likert scales are ordinal, so report top-2-box or the distribution rather than treating the mean as a measurement.
• Instrument data fails informatively: drift biases trends and sensors that cut out at extremes make the most important values missing.

🧪 Practice

1. Construct a sample whose age composition differs from the population, compute weighted and unweighted means, and quantify the bias.
2. Compute a design effect and an adjusted confidence interval, then state the effective sample size in plain language.
3. Interview: A temperature sensor's readings are missing on the ten hottest days of the year. Why is imputing the mean the worst possible choice? (Hint: what determines whether a value is missing?)

<a id="52-data-storage-systems"></a>

5.2 Data Storage Systems

Where data rests determines what you can ask of it and how fast, and the central tension across every system in this section is the same one: optimizing for writing single records conflicts with optimizing for reading many of them.

Relational Database Design and Normalization

Relational design is the discipline of deciding which facts live in which tables. Get it right and correctness is enforced by the database itself; get it wrong and every application that touches the data must remember not to break it.

The problem normalization solves is redundancy. If a customer's address is stored on every one of their orders, then changing the address means updating every row, and any row you miss makes the database disagree with itself. These are the classic three anomalies: update anomalies from partial edits, insertion anomalies when you cannot record a fact without inventing unrelated data, and deletion anomalies when removing a row destroys information it merely happened to carry.

The normal forms are a progression, and the informal summary is exact enough to work from. First normal form requires atomic values, so no comma-separated lists in a cell. Second normal form requires every non-key column to depend on the whole key, which only bites with composite keys. Third normal form requires every non-key column to depend on nothing but the key. The traditional phrasing is that every non-key attribute depends on the key, the whole key, and nothing but the key.

Keys are what make this work. A primary key uniquely identifies a row. A foreign key points at another table's primary key and, when declared, is enforced, so an order cannot reference a customer who does not exist. Declaring constraints moves correctness from convention into the engine.

The counterweight is that normalization costs joins. A fully normalized schema is excellent for transactional systems, where writes are frequent and each touches few rows, and becomes a liability in analytics, where a single question may join eight tables over millions of rows. Warehouses therefore denormalize deliberately, which the next topics cover.

-- ---- THE UNNORMALIZED TABLE, and every anomaly it permits ----
CREATE TABLE orders_bad (
    order_id        INT,
    customer_name   TEXT,          -- repeated on every order
    customer_email  TEXT,          -- repeated, and can disagree between rows
    customer_city   TEXT,
    product_names   TEXT,          -- 'Widget, Gadget'  <- violates 1NF
    order_total     NUMERIC
);
-- UPDATE anomaly:   changing an email means updating N rows; miss one and
--                   the database now holds two emails for one person.
-- INSERTION anomaly: a new customer cannot be recorded until they order.
-- DELETION anomaly:  deleting a customer's only order erases the customer.

-- ---- 1NF: atomic values. Split the repeating group into its own table.
-- ---- 2NF: with a composite key, no partial dependencies.
-- ---- 3NF: no non-key column depends on another non-key column.

CREATE TABLE customers (
    customer_id   SERIAL PRIMARY KEY,
    email         TEXT NOT NULL UNIQUE,      -- a natural key, enforced
    name          TEXT NOT NULL,
    city_id       INT REFERENCES cities(city_id),
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE products (
    product_id    SERIAL PRIMARY KEY,
    sku           TEXT NOT NULL UNIQUE,
    name          TEXT NOT NULL,
    list_price    NUMERIC(10,2) NOT NULL CHECK (list_price >= 0)
);

CREATE TABLE orders (
    order_id      SERIAL PRIMARY KEY,
    customer_id   INT NOT NULL REFERENCES customers(customer_id),
    order_date    DATE NOT NULL,
    status        TEXT NOT NULL
                  CHECK (status IN ('pending','shipped','cancelled'))
);

CREATE TABLE order_items (
    order_id      INT NOT NULL REFERENCES orders(order_id) ON DELETE CASCADE,
    product_id    INT NOT NULL REFERENCES products(product_id),
    quantity      INT NOT NULL CHECK (quantity > 0),
    unit_price    NUMERIC(10,2) NOT NULL,   -- price AT THE TIME OF SALE
    PRIMARY KEY (order_id, product_id)       -- composite natural key
);
-- unit_price is deliberately DUPLICATED from products.list_price. This
-- is not a normalization failure: the price on a historical order is a
-- different fact from the current list price, and it must not change
-- when the catalogue does. Distinguishing "redundant copy" from
-- "point-in-time fact" is the judgement normalization rules cannot make.

-- ---- CONSTRAINTS MOVE CORRECTNESS INTO THE ENGINE ----
-- Without the foreign key, an application bug silently creates an order
-- for customer 99999. With it, the insert fails immediately.
INSERT INTO orders (customer_id, order_date, status)
VALUES (99999, DATE '2024-06-01', 'pending');
-- ERROR: insert or update violates foreign key constraint

-- CHECK constraints encode domain rules the application might forget.
-- UNIQUE prevents the duplicate-key bug at its source.
-- NOT NULL is the cheapest data quality rule available.

-- ---- SLOWLY CHANGING DIMENSIONS: history in a normalized world ----
-- Type 1: overwrite. History is lost.
UPDATE customers SET city_id = 7 WHERE customer_id = 42;

-- Type 2: a new row per version, with validity intervals.
CREATE TABLE customer_history (
    customer_sk   SERIAL PRIMARY KEY,        -- surrogate key, one per VERSION
    customer_id   INT NOT NULL,              -- the stable business key
    city_id       INT,
    valid_from    TIMESTAMPTZ NOT NULL,
    valid_to      TIMESTAMPTZ,               -- NULL = currently in effect
    is_current    BOOLEAN NOT NULL DEFAULT TRUE,
    EXCLUDE USING gist (                      -- no overlapping validity
        customer_id WITH =,
        tstzrange(valid_from, COALESCE(valid_to, 'infinity')) WITH &&
    )
);
-- "What city was this customer in when they placed order 500?" is
-- answerable with Type 2 and unanswerable with Type 1.
SELECT o.order_id, h.city_id
FROM orders o
JOIN customer_history h
  ON h.customer_id = o.customer_id
 AND o.order_date >= h.valid_from::date
 AND o.order_date <  COALESCE(h.valid_to::date, DATE '9999-12-31');

-- ---- NORMALIZED QUERY: correct, and it costs joins ----
SELECT c.name, p.name AS product, oi.quantity, oi.unit_price
FROM orders o
JOIN customers    c  ON c.customer_id = o.customer_id
JOIN order_items  oi ON oi.order_id   = o.order_id
JOIN products     p  ON p.product_id  = oi.product_id
WHERE o.order_date >= DATE '2024-01-01';
-- Four tables for one report. In OLTP this is right. In a warehouse
-- serving a hundred analysts, it is the reason star schemas exist.

  THE THREE ANOMALIES, ON ONE DENORMALIZED TABLE

  order_id  customer_email      city     product
  1         alice@x.com         Paris    Widget
  2         alice@x.com         Paris    Gadget
  3         bob@y.com           Berlin   Widget

  UPDATE    Alice changes email -> 2 rows must change together.
            Update one and the table holds two emails for one person.
  INSERT    A new customer with no orders cannot be recorded at all.
  DELETE    Delete order 3 and Bob's existence disappears with it.

  NORMAL FORMS, informally

  1NF  atomic values            no 'Widget, Gadget' in one cell
  2NF  full key dependency      only relevant with a COMPOSITE key
  3NF  no transitive dependency  city_name must not sit beside city_id

  "the key, the whole key, and nothing but the key"

  WHEN TO STOP
    OLTP        normalize to 3NF; writes are frequent and small
    Analytics   denormalize deliberately; reads dominate and joins cost

Key Takeaways

• Normalization eliminates redundancy and with it the update, insertion, and deletion anomalies that let a database contradict itself.
• Third normal form is reached when every non-key column depends on the key, the whole key, and nothing but the key.
• Declared primary keys, foreign keys, and check constraints move correctness from application convention into the engine.
• A point-in-time fact such as the price at time of sale is not redundant duplication, even though it looks like it.
• Normalization suits transactional workloads and costs joins, which is why analytical schemas denormalize on purpose.

🧪 Practice

1. Take a flat spreadsheet with repeated customer details and decompose it to third normal form, declaring every key and constraint.
2. Design a Type 2 slowly changing dimension and write the query that answers what an attribute's value was on a given past date.
3. Interview: Your schema stores unit_price on order lines although products already have a price. Is this a normalization violation? (Hint: ask whether the two values answer the same question.)

NoSQL: Document, Key-Value, Column, Graph Stores

NoSQL is a poor name for a real idea: relational databases make one set of trade-offs, and some workloads want a different set. The four families below each relax something relational systems guarantee in exchange for something else.

The underlying constraint is the CAP theorem. When a network partition splits the cluster, a distributed system must choose between remaining consistent, refusing to answer rather than risk staleness, and remaining available, answering possibly stale data. Partitions are a fact of networks, so the choice is forced rather than optional. In practice the more useful framing is PACELC: even without a partition, systems trade latency against consistency.

Document stores such as MongoDB hold JSON-like documents with flexible schemas. They suit data whose shape varies per record and that is usually read as a whole. The cost is that cross-document joins are weak, so you denormalize and then own the consistency problem yourself.

Key-value stores such as Redis do one thing: get and set by key, extremely fast. They are caches, session stores, rate limiters, and queues. You cannot query by value.

Wide-column stores such as Cassandra scale writes horizontally and are designed around the queries you will run. The rule is inverted from relational design: you model the table per query, duplicating data freely, because a query the partition key does not serve is effectively impossible.

Graph databases such as Neo4j store relationships as first-class entities. For questions about paths and connectivity, such as fraud rings or recommendation traversals, they express in one line what SQL needs recursive CTEs to approximate, and they stay fast at depth.

# ---- DOCUMENT STORE: flexible shape, whole-object reads ----
# MongoDB. Each document can differ; the schema lives in your code.
order_doc = {
    "_id": "o_987",
    "customer": {                       # EMBEDDED, not a foreign key
        "id": "c_123", "name": "Alice", "tier": "gold",
        "address": {"city": "Paris", "country": "FR"},
    },
    "items": [                          # a nested array, no join table
        {"sku": "A1", "name": "Widget", "qty": 2, "price": 19.99},
        {"sku": "B2", "name": "Gadget", "qty": 1, "price": 24.50},
    ],
    "total": 64.48,
    "placed_at": "2024-06-01T10:00:00Z",
    "gift_message": "Happy birthday",   # present on SOME documents only
}
# One read returns the whole order. No joins, no N+1 query problem.
# The cost: Alice's name is copied into every order she places, so a
# rename requires a multi-document update that is not atomic by default.

# from pymongo import MongoClient
# db = MongoClient()["shop"]
# db.orders.insert_one(order_doc)
# db.orders.create_index([("customer.id", 1), ("placed_at", -1)])
#
# db.orders.find({"customer.tier": "gold", "total": {"$gt": 50}})
#
# An aggregation pipeline is the group-by equivalent:
pipeline = [
    {"$match": {"placed_at": {"$gte": "2024-01-01"}}},      # filter first
    {"$unwind": "$items"},                                   # one row per item
    {"$group": {"_id": "$items.sku",
                "units": {"$sum": "$items.qty"},
                "revenue": {"$sum": {"$multiply":
                            ["$items.qty", "$items.price"]}}}},
    {"$sort": {"revenue": -1}},
    {"$limit": 10},
]
# EMBED when the child is read with the parent and is bounded in size.
# REFERENCE when the child is large, unbounded, or shared across parents.
# An "embed the comments" design fails once one post has 50,000 comments,
# because documents have a size limit and you fetch all of them every read.

# ---- KEY-VALUE STORE: one operation, very fast ----
# import redis
# r = redis.Redis(decode_responses=True)
#
# r.setex("session:abc123", 3600, '{"user_id": 42}')    # with a TTL
# r.get("session:abc123")
#
# CACHE-ASIDE, the pattern that matters for analytics:
def get_metric(key, compute_fn, r=None, ttl=300):
    cached = r.get(key) if r else None
    if cached is not None:
        return float(cached)                 # hit: microseconds
    value = compute_fn()                     # miss: run the expensive query
    if r:
        r.setex(key, ttl, value)
    return value
# A dashboard refreshing every 5 seconds for 200 users runs the warehouse
# query once per TTL rather than 40 times per second.
#
# Other native uses: r.incr("ratelimit:ip:1.2.3.4"), sorted sets for
# leaderboards, r.lpush for a simple queue.
# What you CANNOT do: "find all sessions where user_id = 42". There is
# no index on values. If you need that, this is the wrong store.

# ---- WIDE-COLUMN STORE: model the table per query ----
CASSANDRA_DDL = """
-- Query 1: "recent events for one user, newest first"
CREATE TABLE events_by_user (
    user_id     uuid,
    event_time  timestamp,
    event_id    uuid,
    event_type  text,
    payload     text,
    PRIMARY KEY ((user_id), event_time, event_id)
) WITH CLUSTERING ORDER BY (event_time DESC);
-- (user_id) is the PARTITION key: it decides which node stores the row.
-- event_time is the CLUSTERING key: it decides order WITHIN the partition.

-- Query 2: "all events of one type on one day"
-- This needs a SECOND TABLE. The same data, written twice, on purpose.
CREATE TABLE events_by_type_day (
    event_type  text,
    day         date,
    event_time  timestamp,
    event_id    uuid,
    user_id     uuid,
    PRIMARY KEY ((event_type, day), event_time, event_id)
);
"""
# The relational instinct is one table plus two indexes. In Cassandra
# that produces a query that must contact every node, which does not
# scale and is why ALLOW FILTERING exists as a warning rather than a
# feature. Duplication is the design, not a defect.
#
# The other trap is the HOT PARTITION: partitioning by day alone sends
# every write on a given day to one node. Add a bucket to spread it.

# ---- GRAPH STORE: relationships as first-class ----
CYPHER = """
// Who is within 3 hops of a known fraudulent account, and how?
MATCH path = (f:Account {flagged: true})-[:TRANSFERRED*1..3]->(a:Account)
WHERE a.flagged = false
RETURN a.id AS account,
       length(path) AS hops,
       [n IN nodes(path) | n.id] AS route
ORDER BY hops
LIMIT 20;

// Collaborative filtering in one clause.
MATCH (me:User {id: $uid})-[:BOUGHT]->(:Product)<-[:BOUGHT]-(peer:User),
      (peer)-[:BOUGHT]->(rec:Product)
WHERE NOT (me)-[:BOUGHT]->(rec)
RETURN rec.name, count(DISTINCT peer) AS peers
ORDER BY peers DESC LIMIT 10;
"""
# The variable-length pattern *1..3 is the whole point. In SQL this is a
# recursive CTE whose cost grows sharply with depth, because each hop is
# another join over the full table. A graph store follows pointers, so
# traversal cost depends on the local neighbourhood, not the table size.

Family	Query by	Strength	Give up	Typical use
Relational	anything	joins, constraints, ACID	horizontal write scale	transactions, reporting
Document	key or indexed field	flexible shape, whole-object reads	cross-document joins	catalogues, user profiles, events
Key-value	key only	latency, throughput	any query by value	cache, sessions, rate limits
Wide-column	partition key	write scale, time series	ad-hoc queries	telemetry, IoT, event logs
Graph	pattern match	multi-hop traversal	bulk aggregation	fraud, networks, recommendations

  CAP, AND WHY THE CHOICE IS FORCED

  A network partition splits the cluster:

        node A   X---X   node B
                 (link down)

  A write arrives at A. B cannot see it.

  CONSISTENT (CP)   B refuses to answer rather than serve stale data.
                    Correct, unavailable.   e.g. HBase, etcd, MongoDB (default)
  AVAILABLE  (AP)   B answers with what it has, and reconciles later.
                    Available, possibly stale.  e.g. Cassandra, DynamoDB

  Partitions happen, so you do not choose whether to trade; you choose
  which side to be on when it happens.

  PACELC extends this: Else, with no partition, systems still trade
  Latency against Consistency. A quorum read is slower AND fresher.

Key Takeaways

• Each NoSQL family relaxes a relational guarantee in exchange for scale, latency, or flexibility, so the choice follows from the access pattern.
• Document stores read whole objects cheaply and make you own the consistency of duplicated data.
• Key-value stores query by key only, which makes them ideal caches and useless for filtering by value.
• Wide-column stores are modelled per query, duplicating data across tables because the partition key determines what is answerable.
• Graph stores traverse relationships at a cost set by the neighbourhood rather than the table size, which is what SQL recursion cannot match.

🧪 Practice

1. Model an order with line items as both a normalized relational schema and a single document, and list one query each handles better.
2. Design two Cassandra tables serving different queries over the same events, and explain why one table with two indexes is the wrong answer.
3. Interview: When would you embed a child collection in a document rather than referencing it? (Hint: think about size bounds and whether the child is ever read without the parent.)

Data Warehouses

A data warehouse is a database tuned for analysis rather than transactions. The distinction is not size; it is that the workload is entirely different, and every design decision follows from that.

Transactional systems run OLTP: many small operations, each touching few rows, mostly writes, with latency measured in milliseconds. Warehouses run OLAP: few large queries, each scanning millions of rows, almost entirely reads, aggregating a handful of columns. A schema tuned for the first is actively bad for the second.

The dominant warehouse design is the star schema. A central fact table holds the measurements, one row per event, with foreign keys to dimension tables holding the descriptive attributes. Facts are long and narrow, dimensions are short and wide. A query joins the fact to two or three dimensions, filters on dimension attributes, and aggregates a fact measure.

Star schemas denormalize deliberately. A snowflake schema normalizes the dimensions further, which saves a little space and adds joins, and is usually not worth it. The star exists precisely because analysts should not need eight joins to answer a question.

Grain is the concept to get right first and the one most often mishandled. The grain is what exactly one fact row represents: one order line, one order, one daily summary per store. Every measure must be consistent with the grain, and mixing grains in a single table produces double counting that is extremely hard to detect afterwards.

The other essential distinction is additivity. A measure is additive if summing it across every dimension is meaningful, like revenue. Semi-additive measures such as inventory balance sum across stores but not across time. Non-additive measures such as ratios cannot be summed at all and must be recomputed from their components.

-- ---- STAR SCHEMA ----

-- The DATE dimension: pre-computed so nobody writes date arithmetic twice.
CREATE TABLE dim_date (
    date_key      INT PRIMARY KEY,         -- 20240601, human-readable
    full_date     DATE NOT NULL UNIQUE,
    year          SMALLINT, quarter SMALLINT, month SMALLINT,
    month_name    TEXT, day_of_week SMALLINT, day_name TEXT,
    is_weekend    BOOLEAN, is_holiday BOOLEAN,
    fiscal_year   SMALLINT, fiscal_quarter SMALLINT,
    week_start    DATE
);
-- Fiscal calendars, holidays, and week definitions are business rules.
-- Encoding them ONCE here prevents four teams implementing them four ways.

CREATE TABLE dim_customer (
    customer_key  BIGSERIAL PRIMARY KEY,   -- SURROGATE key: one per version
    customer_id   TEXT NOT NULL,           -- the natural business key
    name          TEXT, email TEXT,
    segment       TEXT, city TEXT, country TEXT, region TEXT,
    -- Type 2 history:
    valid_from    DATE NOT NULL,
    valid_to      DATE NOT NULL DEFAULT DATE '9999-12-31',
    is_current    BOOLEAN NOT NULL DEFAULT TRUE
);
-- The surrogate key is what makes history work: a fact row points at the
-- customer VERSION in force when the event happened, so a later address
-- change does not silently rewrite last year's regional totals.

CREATE TABLE dim_product (
    product_key   BIGSERIAL PRIMARY KEY,
    sku           TEXT NOT NULL,
    name          TEXT, category TEXT, subcategory TEXT, brand TEXT,
    valid_from    DATE, valid_to DATE, is_current BOOLEAN
);

-- The FACT table. GRAIN: exactly one row per order line item.
CREATE TABLE fact_sales (
    sale_key      BIGSERIAL PRIMARY KEY,
    date_key      INT    NOT NULL REFERENCES dim_date(date_key),
    customer_key  BIGINT NOT NULL REFERENCES dim_customer(customer_key),
    product_key   BIGINT NOT NULL REFERENCES dim_product(product_key),
    store_key     BIGINT NOT NULL,
    order_id      TEXT   NOT NULL,         -- degenerate dimension
    -- MEASURES, all at the line-item grain:
    quantity      INT      NOT NULL,
    gross_amount  NUMERIC(12,2) NOT NULL,
    discount      NUMERIC(12,2) NOT NULL DEFAULT 0,
    net_amount    NUMERIC(12,2) NOT NULL,
    cost          NUMERIC(12,2) NOT NULL
);
-- WRONG at this grain: order_shipping_cost. Shipping applies to the
-- ORDER, so storing it per line and summing it multiplies it by the
-- number of lines. Either allocate it across lines or put it in a
-- separate order-grain fact table.

-- ---- THE CANONICAL QUERY: fact joined to dimensions, filtered, aggregated
SELECT
    d.year, d.quarter,
    p.category,
    c.region,
    SUM(f.net_amount)                        AS revenue,
    SUM(f.net_amount - f.cost)               AS margin,
    SUM(f.quantity)                          AS units,
    COUNT(DISTINCT f.order_id)               AS orders,
    -- NON-ADDITIVE: recompute from components, never average the ratio.
    SUM(f.net_amount - f.cost) / NULLIF(SUM(f.net_amount), 0) AS margin_pct
FROM fact_sales f
JOIN dim_date     d ON d.date_key     = f.date_key
JOIN dim_product  p ON p.product_key  = f.product_key
JOIN dim_customer c ON c.customer_key = f.customer_key
WHERE d.year = 2024 AND p.category = 'Electronics'
GROUP BY d.year, d.quarter, p.category, c.region
ORDER BY revenue DESC;

-- WHY margin_pct MUST be computed this way:
--   AVG(line_margin_pct) weights every line equally, so a 1-unit sale
--   with 90% margin offsets a 10,000-unit sale with 5% margin. The
--   ratio of sums is the correct aggregate; the mean of ratios is not.

-- ---- FACT TABLE TYPES ----
-- TRANSACTION fact: one row per event. The most common and most flexible.
-- PERIODIC SNAPSHOT: one row per entity per period. Semi-additive.
CREATE TABLE fact_inventory_daily (
    date_key INT, product_key BIGINT, store_key BIGINT,
    units_on_hand INT,                       -- sums across STORES,
    value_on_hand NUMERIC(12,2)              -- NOT across DAYS
);
-- Summing units_on_hand over a month counts the same stock 30 times.
-- The correct time aggregation is a closing balance or an average.

-- ACCUMULATING SNAPSHOT: one row per process instance, updated in place.
CREATE TABLE fact_order_pipeline (
    order_id TEXT PRIMARY KEY,
    placed_date_key INT, paid_date_key INT,
    shipped_date_key INT, delivered_date_key INT,
    days_to_ship INT, days_to_deliver INT     -- lag measures
);

-- ---- COST CONTROL on a cloud warehouse ----
-- Billing is usually per byte SCANNED, so pruning is money.
SELECT SUM(net_amount) FROM fact_sales
WHERE date_key BETWEEN 20240601 AND 20240630;     -- prunes partitions
-- SELECT * ... scans every column of every partition. On a petabyte
-- table that single query can cost more than a month of compute.

  STAR SCHEMA

                  dim_date
                      |
  dim_customer --- fact_sales --- dim_product
                      |
                  dim_store

  fact    long and narrow    millions to billions of rows, mostly numbers
  dims    short and wide     thousands of rows, mostly text attributes

  one query = fact + 2-3 dims, filter on dim attributes, aggregate a measure

  GRAIN: state it in one sentence before creating the table.
    "one row per order line item"
    "one row per store per day"
  Every measure must be true AT that grain. Mixing grains double counts.

  ADDITIVITY

  additive        revenue, units      sum across ALL dimensions
  semi-additive   balance, headcount  sum across entities, NOT across time
  non-additive    rate, %, average    recompute from components

  The most common warehouse bug: AVG() of a percentage column.

Key Takeaways

• Warehouses serve OLAP workloads of few large scanning queries, which is a different design problem from OLTP's many small writes.
• A star schema puts measurements in a central fact table and descriptive attributes in denormalized dimensions, minimizing joins per question.
• State the fact table's grain in one sentence first, and reject any measure that is not true at that grain.
• Surrogate keys plus validity dates let facts point at the dimension version in force at event time, preserving historical accuracy.
• Non-additive measures such as margin percentage must be recomputed from summed components, never averaged.

🧪 Practice

1. Design a star schema for a subscription business, stating the grain explicitly and classifying each measure as additive, semi-additive, or non-additive.
2. Show numerically why averaging a per-row margin percentage differs from dividing total margin by total revenue.
3. Interview: Someone adds a per-order shipping cost column to a fact table at line-item grain. What breaks, and when will anyone notice? (Hint: consider an order with one line versus one with six.)

Data Lakes and Lakehouses

A warehouse requires you to define the schema before loading anything. That is a strength for governed reporting and an obstacle when you do not yet know what the data is for, which is the situation exploratory work is usually in.

A data lake inverts the order. Store raw files in object storage, cheap and in their original form, and impose structure when you read them. This is schema on read rather than schema on write. It preserves everything, including fields nobody currently needs, which is exactly what you want when next year's question is unknown.

The failure mode is well documented enough to have a name. Without cataloguing, ownership, and quality enforcement, a lake becomes a data swamp: petabytes nobody can find, trust, or interpret. The cost of storage was never the problem; the cost of not knowing what you have is.

The deeper technical problem is that object storage has no transactions. Two writers can corrupt a dataset, a reader can see a half-written update, and there is no way to update or delete individual rows, which makes regulatory deletion requests painful.

The lakehouse solves this with a transaction log over the files. Delta Lake, Apache Iceberg, and Apache Hudi all work the same way: the data stays as Parquet in object storage, and a metadata log records which files constitute each version of the table. That log gives ACID transactions, row-level updates and deletes, schema evolution, and time travel, querying the table as of a past version.

The result is warehouse semantics at lake cost and flexibility, with the caveat that the tooling is younger and query performance still generally trails a purpose-built warehouse on the queries warehouses were tuned for.

# ---- LAKE LAYOUT: zones with different guarantees ----
LAYOUT = """
s3://company-lake/
├── bronze/                      RAW, immutable, exactly as received
│   ├── orders/ingest_date=2024-06-01/part-0000.json.gz
│   └── clickstream/ingest_date=2024-06-01/events.parquet
├── silver/                      CLEANED, typed, deduplicated, conformed
│   └── orders/order_date=2024-06-01/part-0000.parquet
├── gold/                        BUSINESS-READY aggregates and marts
│   └── daily_revenue_by_region/date=2024-06-01/part-0000.parquet
└── _catalog/                    table definitions and lineage
"""
# Bronze is never modified. If a silver transformation turns out to be
# wrong, you reprocess from bronze. Overwriting raw data removes that
# option permanently, which is the one mistake with no recovery.

# ---- WHY PLAIN PARQUET ON OBJECT STORAGE IS NOT ENOUGH ----
PROBLEMS = """
1. NO ATOMICITY   a job writing 500 files crashes after 300.
                  Readers now see a partial dataset with no error.
2. NO ISOLATION   a reader scanning while a writer overwrites sees a mix
                  of old and new files.
3. NO UPDATES     'delete this customer's rows' means rewriting every
                  file that might contain them, by hand.
4. NO SCHEMA MGMT one file adds a column; readers break or silently null.
5. SMALL FILES    streaming writes produce millions of 4 KB files, and
                  listing them costs more than reading them.
"""

# ---- DELTA LAKE: a transaction log over the same Parquet files ----
# from deltalake import DeltaTable, write_deltalake
import pandas as pd
import numpy as np

rng = np.random.default_rng(0)
df = pd.DataFrame({
    "order_id": range(1, 1001),
    "customer_id": rng.integers(1, 100, 1000),
    "amount": rng.random(1000) * 200,
    "order_date": pd.to_datetime("2024-06-01") + pd.to_timedelta(
        rng.integers(0, 30, 1000), "D"),
})

# write_deltalake("s3://lake/silver/orders", df, mode="overwrite",
#                 partition_by=["order_date"])
#
# ATOMIC: the commit is a single log entry. Readers see the old version
# or the new one, never a mixture, even if the writer crashes mid-write.

# ---- UPSERT (MERGE): impossible with plain Parquet, one call here ----
# dt = DeltaTable("s3://lake/silver/orders")
# (dt.merge(source=updates,
#           predicate="target.order_id = source.order_id",
#           source_alias="source", target_alias="target")
#    .when_matched_update_all()
#    .when_not_matched_insert_all()
#    .execute())
#
# This is what makes change data capture and GDPR deletion tractable.
# dt.delete("customer_id = 42")     # row-level delete, logged

# ---- TIME TRAVEL: query the table as it was ----
# dt.load_as_version(3)                          # by version number
# dt.load_as_version(datetime(2024, 6, 1))       # by timestamp
#
# current = DeltaTable(path).to_pandas()
# previous = DeltaTable(path).load_as_version(5).to_pandas()
# changed = current.merge(previous, on="order_id", suffixes=("_now","_then"))
#
# "The dashboard changed overnight and nobody deployed anything" becomes
# a diff between two versions rather than an investigation.

# ---- SCHEMA EVOLUTION, controlled rather than accidental ----
# write_deltalake(path, df_with_new_column, mode="append",
#                 schema_mode="merge")     # add columns explicitly
# Without schema_mode="merge" the append FAILS, which is the right
# default: a silently added column is how downstream code breaks.

# ---- MAINTENANCE: the two operations every lakehouse table needs ----
# dt.optimize.compact()                    # merge small files
# dt.optimize.z_order(["customer_id"])     # co-locate related rows
# dt.vacuum(retention_hours=168)           # delete files older versions used
#
# COMPACTION matters more than it sounds. A streaming job appending every
# minute creates 1,440 files a day. Query time becomes dominated by
# opening files rather than reading data. Compaction routinely gives a
# 10x speedup and is the most commonly skipped maintenance step.
#
# VACUUM permanently removes the files that time travel depends on, so
# the retention period is your time-travel window. Set it deliberately.

# ---- PARTITIONING THE LAKE: the layout decides what is cheap ----
GOOD = "s3://lake/silver/orders/order_date=2024-06-01/region=eu/part-0.parquet"
# Hive-style key=value directories let engines PRUNE: a query filtered to
# one date reads one directory instead of listing the whole table.
BAD_LAYOUTS = """
orders/2024-06-01.parquet          no key=value, no automatic pruning
orders/customer_id=12345/...       too granular: millions of tiny dirs
orders/                            one directory, 4 million files
"""
# Rule of thumb: aim for partitions of 100 MB to 1 GB. Partitioning by a
# high-cardinality column is the most common and most damaging mistake,
# because metadata operations then dominate every query.

# ---- READING A LAKEHOUSE TABLE FROM PANDAS OR DUCKDB ----
# import duckdb
# duckdb.sql("""
#     SELECT region, SUM(amount)
#     FROM delta_scan('s3://lake/silver/orders')
#     WHERE order_date >= '2024-06-01'
#     GROUP BY region
# """)

Property	Warehouse	Data lake	Lakehouse
Schema	on write	on read	on write, evolvable
Storage cost	high	lowest	lowest
Data types	structured	anything	anything, tables typed
Transactions	yes	no	yes
Row-level delete	yes	no	yes
Time travel	limited	no	yes
Query speed	fastest	slowest	close to warehouse
Best for	governed BI	raw retention, ML	both

Key Takeaways

• A lake defers schema to read time, preserving data whose future use is unknown, at the cost of discoverability and trust.
• Plain files on object storage lack atomicity, isolation, row-level updates, and schema management.
• Lakehouse formats add a transaction log over Parquet, giving ACID commits, upserts, deletes, schema evolution, and time travel.
• Keep a raw immutable zone, because overwriting it removes the ability to reprocess when a transformation turns out to be wrong.
• Compaction and partition sizing dominate query performance; partitioning on a high-cardinality column is the most damaging common mistake.

🧪 Practice

1. Write a dataset as both plain Parquet and a Delta table, then simulate an interrupted write and compare what a reader sees in each case.
2. Perform a merge that updates existing rows and inserts new ones, then use time travel to diff the table before and after.
3. Interview: A streaming job appends to a lake table every minute and queries have become ten times slower over three months, with no growth in data volume. What is happening? (Hint: count the files, not the rows.)

Columnar Storage Formats

Columnar storage is the single most impactful idea in analytical data engineering, and its advantage comes from three effects that compound rather than from any one clever trick.

The distinction is about physical layout. Row-oriented storage keeps all fields of a record together, which is what you want when reading or writing whole records, as transactional systems do. Column-oriented storage keeps all values of one field together, which is what you want when reading a few fields across many records, as analytics does.

The first effect is column pruning. A query touching three of fifty columns reads three columns' worth of bytes. A row store must read every row in full to extract three fields, so it does roughly seventeen times the IO.

The second is compression. Values in a column share a type and are often similar, so run-length encoding, dictionary encoding, and delta encoding all work well. A country column with four distinct values compresses to effectively two bits per row. Mixed-type rows in a row store offer far less structure to exploit.

The third is predicate pushdown. Parquet divides data into row groups and stores per-column minimum and maximum statistics for each. A reader can check those statistics and skip an entire row group without decompressing it, which is why sorting data on a frequently filtered column dramatically improves performance.

Parquet and ORC are the on-disk formats; Arrow is the in-memory columnar standard. The pairing matters: Parquet is optimized for storage with heavy compression, Arrow for computation with no decoding step, and modern engines read Parquet directly into Arrow buffers.

import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import os, time

rng = np.random.default_rng(0)
n = 2_000_000
df = pd.DataFrame({
    "event_id": np.arange(n),
    "user_id": rng.integers(1, 50_000, n),
    "country": rng.choice(["FR", "DE", "IT", "ES"], n),    # 4 distinct
    "event_type": rng.choice(["view", "click", "purchase"], n),
    "amount": (rng.random(n) * 100).round(2),
    "ts": pd.to_datetime("2024-01-01") + pd.to_timedelta(np.arange(n), "s"),
})

base = "/tmp/claude-0"
csv_p, pq_p = f"{base}/cols.csv", f"{base}/cols.parquet"
df.to_csv(csv_p, index=False)
df.to_parquet(pq_p, index=False, compression="snappy")

print(f"csv     {os.path.getsize(csv_p)/1e6:8.1f} MB")
print(f"parquet {os.path.getsize(pq_p)/1e6:8.1f} MB")

# ---- EFFECT 1: COLUMN PRUNING ----
t0 = time.perf_counter(); pd.read_csv(csv_p, usecols=["amount"])
print(f"csv, one column     {(time.perf_counter()-t0)*1000:7.0f} ms")
t0 = time.perf_counter(); pd.read_parquet(pq_p, columns=["amount"])
print(f"parquet, one column {(time.perf_counter()-t0)*1000:7.0f} ms")
# CSV must parse every byte of every line to find one field. Parquet
# seeks directly to that column's byte ranges.

# ---- EFFECT 2: COMPRESSION, per column and per codec ----
meta = pq.read_metadata(pq_p)
rg = meta.row_group(0)
print(f"\nrow groups: {meta.num_row_groups}, rows/group: {rg.num_rows:,}")
for i in range(rg.num_columns):
    c = rg.column(i)
    ratio = c.total_uncompressed_size / max(c.total_compressed_size, 1)
    print(f"  {c.path_in_schema:<12} {c.compression:<7} "
          f"{c.total_compressed_size/1e6:6.2f} MB  {ratio:5.1f}x  "
          f"{c.encodings}")
# `country` compresses enormously: dictionary encoding stores 4 strings
# once plus 2-bit codes. `event_id` compresses well too, via delta
# encoding on a monotonic sequence. `amount` barely compresses, because
# random floats have no structure to exploit.

for codec in ["snappy", "zstd", "gzip", None]:
    p = f"{base}/c_{codec}.parquet"
    t0 = time.perf_counter()
    df.to_parquet(p, compression=codec, index=False)
    wt = time.perf_counter() - t0
    t0 = time.perf_counter(); pd.read_parquet(p); rt = time.perf_counter() - t0
    print(f"{str(codec):<7} {os.path.getsize(p)/1e6:7.1f} MB  "
          f"write {wt:5.2f}s  read {rt:5.2f}s")
# snappy: fast, moderate ratio, the sensible default.
# zstd:   ~30% smaller than snappy at similar read speed. Best for cold
#         data or when egress is billed.
# gzip:   smallest, slow. Only for archival.

# ---- EFFECT 3: PREDICATE PUSHDOWN via row-group statistics ----
sorted_p = f"{base}/sorted.parquet"
df.sort_values("ts").to_parquet(sorted_p, index=False, row_group_size=100_000)

m = pq.read_metadata(sorted_p)
ts_idx = m.schema.names.index("ts")
for g in range(3):
    st = m.row_group(g).column(ts_idx).statistics
    print(f"group {g}: {st.min} .. {st.max}")
# Because the data is SORTED, each row group covers a distinct time
# range, so a filtered query can skip most groups outright.

cutoff = pd.Timestamp("2024-01-20")
t0 = time.perf_counter()
hit = pd.read_parquet(sorted_p, filters=[("ts", ">=", cutoff)])
print(f"sorted   : {len(hit):,} rows in {(time.perf_counter()-t0)*1000:.0f} ms")

shuffled_p = f"{base}/shuffled.parquet"
df.sample(frac=1, random_state=0).to_parquet(
    shuffled_p, index=False, row_group_size=100_000)
t0 = time.perf_counter()
hit2 = pd.read_parquet(shuffled_p, filters=[("ts", ">=", cutoff)])
print(f"shuffled : {len(hit2):,} rows in {(time.perf_counter()-t0)*1000:.0f} ms")
# Identical data, identical result, very different time. When rows are
# shuffled every row group spans the full range, so no group can be
# skipped. SORT ON YOUR MOST-FILTERED COLUMN BEFORE WRITING.

# ---- ROW GROUP SIZE is a real tuning knob ----
for size in [10_000, 128 * 1024, 1_000_000]:
    p = f"{base}/rg_{size}.parquet"
    df.to_parquet(p, row_group_size=size, index=False)
    mm = pq.read_metadata(p)
    print(f"row_group_size {size:>9,}: {mm.num_row_groups:>4} groups, "
          f"{os.path.getsize(p)/1e6:6.1f} MB")
# Too small: metadata overhead and poor compression.
# Too large: coarse pruning and high memory per read.
# 128 MB of uncompressed data per group is a reasonable target.

# ---- ARROW: the in-memory counterpart ----
table = pq.read_table(pq_p)
print(f"\narrow table: {table.num_rows:,} rows, {table.nbytes/1e6:.0f} MB")
print(table.schema)
# Arrow columns are already the layout compute engines want, so reading
# Parquet into Arrow involves decompression but no row-to-column
# transposition, and handing the table to DuckDB, Polars, or Spark is a
# pointer pass rather than a serialization round trip.

# ---- PARTITIONED DATASETS: pruning at the directory level ----
import pyarrow.dataset as ds
part_dir = f"{base}/partitioned"
pq.write_to_dataset(table, root_path=part_dir,
                    partition_cols=["country", "event_type"])
dataset = ds.dataset(part_dir, format="parquet", partitioning="hive")
subset = dataset.to_table(
    filter=(ds.field("country") == "FR") & (ds.field("event_type") == "purchase"),
    columns=["user_id", "amount"],
)
print(f"{subset.num_rows:,} rows from 2 of 12 directories")
# Two levels of pruning now apply: directories eliminated by the
# partition filter, then row groups eliminated by column statistics.

  ROW VERSUS COLUMN LAYOUT

  logical table
    id   country  amount   ts
    1    FR       19.99    10:00
    2    DE       24.50    10:01
    3    FR       15.00    10:02

  ROW STORE on disk
    [1|FR|19.99|10:00][2|DE|24.50|10:01][3|FR|15.00|10:02]
    SELECT SUM(amount) must touch every byte of every record.

  COLUMN STORE on disk
    id:      [1][2][3]
    country: [FR][DE][FR]        -> dictionary: {0:FR,1:DE} + [0][1][0]
    amount:  [19.99][24.50][15.00]
    ts:      [10:00][10:01][10:02] -> delta: [10:00][+1][+1]

    SELECT SUM(amount) reads ONE contiguous run of typed values.

  THREE COMPOUNDING EFFECTS
    prune     read 3 of 50 columns        ~17x less IO
    compress  typed, similar neighbours   ~5-20x smaller
    skip      min/max per row group       read only matching groups

  PREDICATE PUSHDOWN NEEDS SORTED DATA

  sorted by ts                    shuffled
  group 0: 10:00 .. 10:59         group 0: 10:00 .. 23:59
  group 1: 11:00 .. 11:59         group 1: 10:00 .. 23:59
  group 2: 12:00 .. 12:59         group 2: 10:00 .. 23:59

  WHERE ts >= 12:00 skips         no group can be skipped: every group
  groups 0 and 1 entirely         might contain a match

Key Takeaways

• Column layout wins through three compounding effects: reading only the columns needed, compressing typed neighbours, and skipping row groups.
• Dictionary, run-length, and delta encodings make low-cardinality and monotonic columns compress by an order of magnitude or more.
• Predicate pushdown only helps when data is sorted on the filtered column, because shuffled row groups all span the full range.
• Row group size trades pruning granularity against metadata overhead; around 128 MB uncompressed is a sensible target.
• Parquet is the storage format and Arrow the in-memory one, and the pair lets engines exchange data without serialization.

🧪 Practice

1. Write the same two-million-row frame as CSV and Parquet, then time reading a single column from each and explain the ratio.
2. Write one Parquet file sorted on a timestamp and one shuffled, then compare filtered read times and inspect the row group statistics.
3. Interview: Two Parquet files hold identical data and identical schemas, but one is three times larger. Name two plausible causes. (Hint: consider the codec and the ordering of rows.)

Partitioning and Indexing Strategies

Partitioning and indexing are the two mechanisms for avoiding reading data you do not need. They work at different scales and solve different problems, and using one where the other belongs is a common source of disappointing performance.

Partitioning physically splits a table into pieces by the value of a column. A query filtered on that column reads only the matching pieces, a process called partition pruning, and everything else is skipped without being opened. Partitioning is coarse: it eliminates large chunks cheaply.

Indexing builds a separate sorted structure mapping values to row locations. A query filtered on an indexed column looks up the value and jumps to the matching rows. Indexing is fine-grained: it finds a few rows among many.

The rule of thumb follows from that difference. Partition on a low-cardinality, frequently-filtered column, almost always a date. Index on a high-cardinality column used for point lookups and joins, such as an identifier.

The most damaging mistake is partitioning on a high-cardinality column. Partitioning by customer id creates a directory per customer, and the cost of listing and opening millions of tiny files exceeds any benefit by a large margin. The symptom is a query that spends nearly all its time in metadata operations.

Indexes are not free either. Each one must be updated on every write and occupies storage, so an over-indexed table has slow inserts. The discipline is to add indexes from observed query patterns rather than speculation, and to remove unused ones.

For warehouses and lakehouses, clustering or Z-ordering is the third option: physically co-locating rows with similar values across several columns at once, which improves row-group skipping without creating directories.

-- ---- PARTITIONING a relational table by range ----
CREATE TABLE events (
    event_id   BIGSERIAL,
    user_id    BIGINT NOT NULL,
    event_type TEXT   NOT NULL,
    occurred_at TIMESTAMPTZ NOT NULL,
    payload    JSONB
) PARTITION BY RANGE (occurred_at);

CREATE TABLE events_2024_05 PARTITION OF events
    FOR VALUES FROM ('2024-05-01') TO ('2024-06-01');
CREATE TABLE events_2024_06 PARTITION OF events
    FOR VALUES FROM ('2024-06-01') TO ('2024-07-01');

-- PRUNING: this reads ONE partition, not the table.
EXPLAIN SELECT COUNT(*) FROM events
WHERE occurred_at >= '2024-06-01' AND occurred_at < '2024-07-01';
--   -> Seq Scan on events_2024_06     (events_2024_05 never opened)

-- PRUNING FAILS when the partition key is wrapped in a function.
-- WHERE date_trunc('month', occurred_at) = '2024-06-01'
--   -> scans EVERY partition. Always filter on the raw column.

-- DROPPING OLD DATA becomes instant metadata, not a slow DELETE.
DROP TABLE events_2024_05;       -- vs DELETE ... WHERE occurred_at < ...
-- A DELETE of 500 million rows writes 500 million undo records, holds
-- locks, and bloats the table. DROP PARTITION is one catalogue update.
-- This alone justifies partitioning on any table with a retention policy.

-- ---- INDEXES: the kinds, and when each applies ----
CREATE INDEX idx_events_user       ON events (user_id);          -- B-tree
CREATE INDEX idx_events_user_time  ON events (user_id, occurred_at DESC);
CREATE INDEX idx_events_payload    ON events USING gin (payload); -- JSONB
CREATE INDEX idx_events_recent     ON events (user_id)
    WHERE occurred_at > '2024-06-01';                             -- PARTIAL
CREATE INDEX idx_events_lower_type ON events (lower(event_type)); -- EXPRESSION

-- PARTIAL indexes are undervalued: if 95% of queries touch recent rows,
-- an index over only those rows is a fraction of the size and stays
-- resident in memory.

-- COMPOSITE INDEX COLUMN ORDER: equality first, range second.
-- idx (user_id, occurred_at) serves:
--   WHERE user_id = 7                              yes
--   WHERE user_id = 7 AND occurred_at > '...'      yes, fully
--   WHERE occurred_at > '...'                      NO: not a left prefix

-- COVERING INDEX: answer from the index alone, never touching the table.
CREATE INDEX idx_cover ON events (user_id, occurred_at)
    INCLUDE (event_type);

-- ---- FINDING UNUSED AND MISSING INDEXES ----
SELECT relname, indexrelname, idx_scan, pg_size_pretty(
           pg_relation_size(indexrelid)) AS size
FROM pg_stat_user_indexes
WHERE idx_scan < 50                      -- scanned almost never
ORDER BY pg_relation_size(indexrelid) DESC;
-- Every one of these costs write throughput and storage for no benefit.

SELECT relname, seq_scan, seq_tup_read, idx_scan
FROM pg_stat_user_tables
WHERE seq_scan > idx_scan AND seq_tup_read > 1e6
ORDER BY seq_tup_read DESC;              -- candidates for a new index

# ---- PARTITIONING A LAKE: the cardinality decision ----
import pandas as pd, numpy as np, os, shutil
import pyarrow as pa, pyarrow.parquet as pq

rng = np.random.default_rng(0)
n = 500_000
df = pd.DataFrame({
    "user_id": rng.integers(1, 100_000, n),            # HIGH cardinality
    "country": rng.choice(["FR", "DE", "IT"], n),      # LOW cardinality
    "day": pd.to_datetime("2024-06-01") + pd.to_timedelta(
        rng.integers(0, 30, n), "D"),                  # LOW, and filtered
    "amount": rng.random(n) * 100,
})
table = pa.Table.from_pandas(df, preserve_index=False)

def build(cols, label):
    path = f"/tmp/claude-0/part_{label}"
    shutil.rmtree(path, ignore_errors=True)
    pq.write_to_dataset(table, root_path=path, partition_cols=cols)
    files = sum(len(f) for _, _, f in os.walk(path))
    size = sum(os.path.getsize(os.path.join(r, f))
               for r, _, fs in os.walk(path) for f in fs)
    print(f"{label:<18} {files:>7,} files  {size/1e6:7.1f} MB  "
          f"{size/max(files,1)/1024:8.1f} KB/file")

build(["day"], "by day")                       # 30 partitions: good
build(["country", "day"], "country+day")       # 90 partitions: still fine
build(["user_id"], "by user_id")               # ~100,000 partitions: BAD
# The last layout produces files of a few kilobytes each. Listing and
# opening them dominates every query, and compression is destroyed
# because each file is too small for the encoder to find patterns.

# THE SIZING RULE: target 100 MB to 1 GB per partition.
def suggest(df, col, target_mb=256):
    total_mb = df.memory_usage(deep=True).sum() / 1e6
    parts = df[col].nunique()
    per = total_mb / parts
    verdict = ("too granular" if per < 50 else
               "too coarse" if per > 2000 else "reasonable")
    print(f"{col:<10} {parts:>7,} partitions  ~{per:8.2f} MB each  {verdict}")

for c in ["day", "country", "user_id"]:
    suggest(df, c)

# ---- WHEN THE NATURAL KEY IS TOO GRANULAR: bucket it ----
df["user_bucket"] = df["user_id"] % 64        # 64 even buckets, not 100,000
# A query for one user reads 1/64 of the data instead of all of it, and
# the file count stays manageable. This is hash partitioning.

# ---- CLUSTERING / Z-ORDER: skipping without directories ----
# Sorting co-locates similar values so row-group statistics become tight.
sorted_df = df.sort_values(["day", "country"])
pq.write_table(pa.Table.from_pandas(sorted_df, preserve_index=False),
               "/tmp/claude-0/clustered.parquet", row_group_size=50_000)
# Delta / Iceberg expose this directly:
#   dt.optimize.z_order(["country", "user_id"])
# Z-ordering interleaves the bits of several columns so that filtering on
# ANY of them prunes well, which plain sorting only achieves for the
# leading column.

# ---- VERIFY PRUNING ACTUALLY HAPPENS ----
import pyarrow.dataset as ds
dataset = ds.dataset("/tmp/claude-0/part_by day", partitioning="hive")
scanned = dataset.to_table(filter=ds.field("day") == "2024-06-15")
print(f"{scanned.num_rows:,} of {len(df):,} rows scanned")
# If this number equals the full row count, pruning is NOT happening,
# and the usual cause is a filter expressed against a derived column.

Mechanism	Granularity	Best on	Cost
Partitioning	whole chunks	low-cardinality, always-filtered (date)	tiny files if cardinality is high
B-tree index	individual rows	high-cardinality lookups and joins	write overhead, storage
Partial index	subset of rows	a hot subset such as recent data	one more object to maintain
Clustering / Z-order	row groups	several filtered columns	periodic re-optimization
Bucketing	fixed chunks	high-cardinality joins	fixed bucket count

Key Takeaways

• Partitioning eliminates large chunks by directory or range, while indexing locates individual rows; the two solve different problems.
• Partition on a low-cardinality, frequently-filtered column, almost always a date, and index high-cardinality lookup columns.
• Partitioning on a high-cardinality column creates millions of tiny files whose metadata cost exceeds any pruning benefit.
• Wrapping the partition key in a function defeats pruning, exactly as it defeats index use.
• Clustering and Z-ordering improve row-group skipping across several columns without creating directories.

🧪 Practice

1. Write the same dataset partitioned by date and by a high-cardinality identifier, then compare file counts, total size, and query time.
2. Create a composite index and demonstrate one query that uses it and one that cannot, explaining the leftmost prefix rule.
3. Interview: A partitioned table's queries are slow and the plan shows every partition being read despite a filter on the partition column. What would you look at? (Hint: is the filter applied to the column itself?)

<a id="53-data-pipelines"></a>

5.3 Data Pipelines

A pipeline is the code that moves data from where it is produced to where it is analysed, and because it runs unattended on data that changes, its design problem is not correctness once but correctness on every future run.

ETL vs ELT

The three letters stand for extract, transform, and load, and the only question is where the T happens. That ordering decision determines your cost structure, your flexibility, and who can change the logic.

ETL transforms before loading. Data is extracted, processed on a dedicated machine, and the finished result is written to the warehouse. This was the standard for decades, for a concrete reason: warehouse storage and compute were expensive and tightly coupled, so you loaded only what you needed in exactly the shape you needed it.

ELT loads raw data first and transforms inside the warehouse. Cloud warehouses separated storage from compute and made storage cheap, which removed the reason to filter before loading. Transformations become SQL against tables that already exist, which means analysts can write them and version them like any other code.

The practical advantages of ELT are worth being specific about. Raw data is retained, so a transformation bug is fixable by rerunning SQL rather than re-extracting from a source that may no longer hold the history. The warehouse's query engine is usually faster than a single Python process. And the transformation layer becomes SQL, which more of the team can read.

ETL still wins in identifiable situations: when personally identifiable data must be masked or dropped before it lands, when the source volume is far larger than what you need, when the transformation is genuinely not expressible in SQL, and when egress or warehouse compute costs dominate.

The modern default is ELT with a medallion structure, bronze for raw, silver for cleaned and conformed, gold for business-ready aggregates, each layer a set of SQL models with tests.

# ---- ETL: transform in Python, load the finished result ----
import pandas as pd
import numpy as np
from sqlalchemy import create_engine, text

def etl_pipeline(source_engine, warehouse_engine, run_date):
    # EXTRACT: pull only what is needed, filtered at the source.
    raw = pd.read_sql(
        text("""SELECT order_id, customer_id, amount, status, created_at
                FROM orders WHERE created_at::date = :d"""),
        source_engine, params={"d": run_date},
    )

    # TRANSFORM: all business logic lives in this Python function.
    clean = (
        raw.loc[raw["status"] != "test"]
           .assign(
               amount=lambda d: d["amount"].clip(lower=0),
               created_at=lambda d: pd.to_datetime(d["created_at"], utc=True),
               is_large=lambda d: d["amount"] > 1000,
           )
           .drop_duplicates(subset=["order_id"], keep="last")
    )

    # LOAD: only the finished table reaches the warehouse.
    clean.to_sql("fact_orders", warehouse_engine, if_exists="append",
                 index=False, method="multi")
    return len(clean)

# The cost: the raw rows are gone. If `status != 'test'` turns out to
# have excluded a legitimate status, the only recovery is re-extracting
# from the source, which may have been purged or may have changed.

# ---- ELT: load raw, transform with SQL in the warehouse ----
def el_pipeline(source_engine, warehouse_engine, run_date):
    """Extract and Load only. No business logic whatsoever."""
    raw = pd.read_sql(
        text("SELECT * FROM orders WHERE created_at::date = :d"),
        source_engine, params={"d": run_date},
    )
    raw["_ingested_at"] = pd.Timestamp.now(tz="UTC")
    raw["_source"] = "oltp.orders"
    raw.to_sql("raw_orders", warehouse_engine, schema="bronze",
               if_exists="append", index=False, method="multi")
    return len(raw)
# Note SELECT *: in ELT you take everything, including columns nobody
# currently uses, because storage is cheap and re-extraction is not.

-- ---- THE T IN ELT: SQL models, versioned like any other code ----

-- silver: cleaned, typed, deduplicated. One row per order.
CREATE OR REPLACE VIEW silver.orders AS
WITH deduplicated AS (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY order_id
                              ORDER BY _ingested_at DESC) AS rn
    FROM bronze.raw_orders
)
SELECT
    order_id,
    customer_id,
    GREATEST(amount, 0)                       AS amount,
    LOWER(TRIM(status))                       AS status,
    created_at AT TIME ZONE 'UTC'             AS created_at_utc,
    amount > 1000                             AS is_large
FROM deduplicated
WHERE rn = 1                                   -- keep the latest version
  AND status <> 'test';
-- The exclusion of 'test' is now a line of SQL in version control.
-- Changing it and rebuilding costs one query; it does not require the
-- source system to still hold last year's rows.

-- gold: the business-ready aggregate the dashboard reads.
CREATE OR REPLACE VIEW gold.daily_revenue AS
SELECT
    created_at_utc::date                      AS day,
    COUNT(*)                                  AS orders,
    COUNT(DISTINCT customer_id)               AS customers,
    SUM(amount)                               AS revenue,
    SUM(amount) FILTER (WHERE is_large)       AS large_order_revenue
FROM silver.orders
GROUP BY 1;

# ---- dbt: the standard way to manage the T in ELT ----
# models/silver/orders.sql holds the SELECT above, and this declares
# what must be true about its output.
version: 2
models:
  - name: orders
    description: "One row per order, deduplicated and typed."
    columns:
      - name: order_id
        tests: [unique, not_null]          # the grain, enforced
      - name: customer_id
        tests:
          - not_null
          - relationships:                 # referential integrity
              to: ref('customers')
              field: customer_id
      - name: amount
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              inclusive: true
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'shipped', 'cancelled']
# Tests run on every build. A source adding a new status value fails the
# pipeline loudly instead of silently appearing as an unhandled category
# three dashboards downstream.

  WHERE THE T HAPPENS

  ETL                                  ELT
  source                               source
    |  extract                           |  extract
    v                                    v
  transform server  <- your code       warehouse (bronze)  <- raw, all of it
    |  clean, filter, reshape            |  transform, in SQL
    v                                    v
  warehouse         <- final only      warehouse (silver -> gold)

  ETL                                  ELT
  + less data stored and moved         + raw retained: reprocess anytime
  + PII removed before landing         + warehouse compute does the work
  + arbitrary Python transformations   + SQL: reviewable by more of the team
  - raw is gone; bugs need re-extract  - stores data you may never use
  - logic hidden in a Python repo      - PII lands in the warehouse first
  - transform server is a bottleneck   - warehouse compute is billed

  CHOOSE ETL WHEN: PII must not land, source volume >> needed volume,
                   the transform is not expressible in SQL.
  OTHERWISE:       ELT with bronze / silver / gold.

Key Takeaways

• The only difference is where transformation happens, and that choice sets your cost structure and who can change the logic.
• ELT retains raw data, so a transformation bug is fixed by rerunning SQL rather than re-extracting from a source that may have moved on.
• Cloud warehouses separating storage from compute is what made loading raw data first economically sensible.
• ETL remains correct when sensitive data must never land, when source volume vastly exceeds what is needed, or when the logic is not SQL.
• Structure ELT as bronze, silver, and gold layers with tests attached to each model.

🧪 Practice

1. Implement the same cleaning logic as an ETL Python function and as an ELT SQL view, then describe what recovery looks like in each when the filter turns out to be wrong.
2. Write schema tests asserting uniqueness of the grain, a value range, and an accepted value set, then feed in violating data.
3. Interview: Your company handles data subject to deletion requests and is considering ELT. What complication does raw retention introduce? (Hint: where does the data now exist that it did not before?)

Batch vs Streaming Ingestion

Batch and streaming are two answers to the question of when data moves. The choice is usually made by habit, and it should be made by looking at what the consumer actually needs.

Batch processes a bounded set of records on a schedule. It is simpler in every dimension: easy to reason about, easy to test with a fixed input, easy to reprocess by rerunning, and cheap because resources are used only while the job runs. The cost is latency equal to roughly half the interval on average.

Streaming processes records continuously as they arrive. Latency drops to seconds, and in exchange you accept permanently running infrastructure, out-of-order arrival, state that must be maintained across events, and the fact that reprocessing history requires a replay mechanism.

The decision rule is uncomfortable but correct: streaming is justified only when a decision is made faster than the batch interval. Fraud blocking, dynamic pricing, and operational alerting qualify. A dashboard that executives read each morning does not, no matter how much fresher streaming would make it feel.

Between the two sits micro-batching, which runs small batches every few minutes. It captures most of the latency benefit with most of the simplicity of batch, and it is the right answer more often than either pure option.

The architectural patterns are worth naming. Lambda runs a batch layer and a streaming layer in parallel, giving accuracy and speed at the cost of maintaining the same logic twice, which inevitably drifts. Kappa uses one streaming path and reprocesses by replaying the log, which is simpler but requires the log to retain enough history.

import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

# ---- BATCH: bounded input, scheduled, idempotent by construction ----
def batch_load(source, warehouse, window_start, window_end):
    """Process one closed time window. Rerunnable without side effects."""
    rows = source.read(
        f"SELECT * FROM events "
        f"WHERE occurred_at >= '{window_start}' AND occurred_at < '{window_end}'"
    )
    # HALF-OPEN INTERVAL [start, end). A closed interval on both ends
    # double counts every boundary record in adjacent runs.

    aggregated = (
        rows.groupby(["region", pd.Grouper(key="occurred_at", freq="h")])
            .agg(events=("event_id", "count"), revenue=("amount", "sum"))
            .reset_index()
    )
    # DELETE-THEN-INSERT for the window makes a rerun safe.
    warehouse.execute(
        "DELETE FROM hourly_metrics WHERE hour >= %s AND hour < %s",
        (window_start, window_end),
    )
    warehouse.insert("hourly_metrics", aggregated)
    return len(aggregated)

# ---- MICRO-BATCH: the middle ground, and often the right answer ----
def micro_batch_loop(source, warehouse, interval_s=300, lag_s=60):
    """Every 5 minutes, process the window ending 60 seconds ago."""
    while True:
        now = datetime.now(timezone.utc)
        end = now - timedelta(seconds=lag_s)       # allow for arrival lag
        start = end - timedelta(seconds=interval_s)
        batch_load(source, warehouse, start, end)
        # The lag is the key parameter: processing right up to `now`
        # misses records still in flight, which then never get counted.
        break                                       # illustrative

# ---- STREAMING: unbounded input, continuous, stateful ----
class StreamingAggregator:
    """Windowed aggregation with late-arrival handling."""

    def __init__(self, window_s=300, allowed_lateness_s=120):
        self.window_s = window_s
        self.lateness = allowed_lateness_s
        self.state = {}                 # window_start -> running aggregate
        self.max_event_time = None
        self.dropped_late = 0

    def _window_of(self, ts):
        return int(ts.timestamp()) // self.window_s * self.window_s

    def process(self, event):
        ts = event["occurred_at"]
        self.max_event_time = ts if self.max_event_time is None \
            else max(self.max_event_time, ts)
        watermark = self.max_event_time - timedelta(seconds=self.lateness)

        w = self._window_of(ts)
        w_end = datetime.fromtimestamp(w + self.window_s, timezone.utc)
        if w_end < watermark:
            self.dropped_late += 1      # COUNT these; a rising count is a signal
            return None

        agg = self.state.setdefault(w, {"count": 0, "sum": 0.0})
        agg["count"] += 1
        agg["sum"] += event["amount"]
        return w

    def emit_closed(self):
        """Windows entirely behind the watermark are final and can ship."""
        wm = self.max_event_time - timedelta(seconds=self.lateness)
        ready = [w for w in self.state
                 if datetime.fromtimestamp(w + self.window_s, timezone.utc) < wm]
        out = [{"window_start": datetime.fromtimestamp(w, timezone.utc),
                **self.state.pop(w)} for w in ready]
        return out

rng = np.random.default_rng(0)
base = datetime(2024, 6, 1, 10, 0, tzinfo=timezone.utc)
agg = StreamingAggregator()
for i in range(200):
    lateness = timedelta(seconds=int(rng.exponential(30)))   # skewed arrival
    agg.process({"occurred_at": base + timedelta(seconds=i * 3) - lateness,
                 "amount": float(rng.random() * 100)})
print(f"open windows {len(agg.state)}, dropped late {agg.dropped_late}")
print(agg.emit_closed())

# THE STATE IS THE HARD PART. It lives in memory, so a crash loses it
# unless checkpointed, and it grows without bound unless windows are
# closed and evicted. Batch has neither problem, which is the real
# reason batch is simpler rather than merely less fashionable.

# ---- CHOOSING, as a function rather than a preference ----
def recommend(decision_latency_s, volume_per_day, team_size, needs_replay):
    if decision_latency_s < 60:
        return "streaming: a decision is made faster than any batch interval"
    if decision_latency_s < 900:
        return "micro-batch: most of the latency benefit, batch simplicity"
    if volume_per_day > 1e9 and team_size >= 4:
        return "streaming: volume makes a single batch window impractical"
    return "batch: the simplest thing that meets the requirement"

for case in [
    ("fraud blocking", 5, 1e7, 6, False),
    ("ops alerting", 120, 1e8, 4, False),
    ("exec dashboard", 43200, 1e7, 2, True),
    ("monthly finance close", 2.6e6, 1e6, 2, True),
]:
    name, *args = case
    print(f"{name:<22} {recommend(*args)}")

  LATENCY VERSUS COMPLEXITY

  daily batch      latency ~12h avg   one cron job, one SQL script
  hourly batch     latency ~30m avg   same code, different schedule
  micro-batch 5m   latency ~2.5m      same code, plus a lag parameter
  streaming        latency ~seconds   state, watermarks, checkpoints,
                                      replay, always-on infrastructure

  Each step down multiplies operational cost. Take a step only when a
  DECISION depends on the improvement.

  THE QUESTION THAT SETTLES IT
    "What action is taken differently if this arrives in 5 seconds
     instead of 4 hours?"
    No answer -> batch.

  LAMBDA VS KAPPA

  LAMBDA                          KAPPA
  source --> batch layer  --\     source --> stream layer --> serving
         \-> stream layer --+--> serving
                                  one code path
  two code paths computing        reprocess by REPLAYING the log
  the same metric; they drift,    requires log retention long enough
  and reconciling them is a       to cover a full rebuild
  permanent tax

Key Takeaways

• Batch processes bounded input on a schedule and is simpler to test, reason about, and reprocess.
• Streaming buys seconds of latency in exchange for state management, out-of-order handling, and always-on infrastructure.
• Justify streaming by a decision made faster than the batch interval, not by a preference for fresher dashboards.
• Micro-batching captures most of the latency benefit while keeping batch simplicity, and is underused.
• Use half-open time windows, and process up to a lag behind now, or in-flight records are silently never counted.

🧪 Practice

1. Implement a batch aggregation over a half-open window and demonstrate that running it twice produces identical output.
2. Build a micro-batch loop with a configurable lag, then show what happens to record counts when the lag is set to zero.
3. Interview: A stakeholder asks for a real-time dashboard. What is the one question you ask before agreeing? (Hint: it concerns what happens differently as a result of the freshness.)

Workflow Orchestration

A pipeline is rarely one script. It is a set of tasks with dependencies: extract three sources, join them, build features, train a model, publish a dashboard. An orchestrator manages running those tasks in the right order, at the right time, and handling what happens when one fails.

The naive alternative is a chain of cron jobs, and it fails in predictable ways. Cron has no concept of dependencies, so job B runs at its scheduled time whether or not job A succeeded. It has no retries, no visibility into which step failed, no way to rerun one step, and no record of past runs.

Orchestrators model work as a DAG, a directed acyclic graph. Nodes are tasks and edges are dependencies. Acyclic matters: a cycle would mean a task depends on itself, which cannot be scheduled. The DAG lets the scheduler determine what can run in parallel and what must wait.

The other core concept is the execution date, which is not the same as the wall-clock time the task runs. A daily job for 1 June typically runs early on 2 June, once the day is complete. Tasks should be parameterized by that logical date rather than by now(), which is what makes backfills possible: running the pipeline over a range of past dates by supplying different logical dates to the same code.

Two philosophies exist. Task-centric tools such as Airflow model "run this, then that". Asset-centric tools such as Dagster model "this table should exist and be fresh", deriving execution from the dependency graph between data assets. The second matches how analysts think about data, and the shift toward it is ongoing.

# ---- AIRFLOW: task-centric orchestration ----
from datetime import datetime, timedelta
# from airflow import DAG
# from airflow.decorators import dag, task

DEFAULT_ARGS = {
    "owner": "analytics",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "email_on_failure": True,
    "execution_timeout": timedelta(hours=2),   # kill a hung task
    "depends_on_past": False,                  # True serializes the backfill
}

# @dag(
#     dag_id="daily_revenue",
#     start_date=datetime(2024, 1, 1),
#     schedule="0 3 * * *",              # 03:00 daily
#     catchup=False,                     # do NOT auto-run every missed day
#     max_active_runs=1,                 # no two runs of this DAG at once
#     default_args=DEFAULT_ARGS,
#     tags=["revenue", "daily"],
# )
def daily_revenue():
    # @task
    def extract_orders(ds: str) -> str:
        """ds is the LOGICAL date, injected by the scheduler."""
        # Parameterizing by ds rather than date.today() is what makes
        # this task rerunnable for any past day.
        path = f"s3://lake/bronze/orders/ingest_date={ds}/"
        return path

    # @task
    def extract_refunds(ds: str) -> str:
        return f"s3://lake/bronze/refunds/ingest_date={ds}/"

    # @task
    def validate(paths: list[str], ds: str) -> dict:
        """Fail the run here rather than publishing bad numbers."""
        checks = {"row_count_ok": True, "no_nulls_in_key": True}
        if not all(checks.values()):
            raise ValueError(f"validation failed for {ds}: {checks}")
        return checks

    # @task
    def transform(ds: str) -> str:
        return f"s3://lake/silver/revenue/date={ds}/"

    # @task
    def publish(ds: str) -> None:
        ...

    # DEPENDENCIES are inferred from how the outputs are passed.
    orders = extract_orders("{{ ds }}")
    refunds = extract_refunds("{{ ds }}")      # runs in PARALLEL with orders
    checked = validate([orders, refunds], "{{ ds }}")
    built = transform("{{ ds }}")
    publish("{{ ds }}")

# CATCHUP is the setting that surprises everyone once: with catchup=True
# and a start_date six months back, deploying the DAG immediately
# schedules ~180 runs. Leave it False and backfill deliberately:
#   airflow dags backfill daily_revenue -s 2024-01-01 -e 2024-03-31

# ---- DAGSTER: asset-centric ----
# from dagster import asset, AssetIn, DailyPartitionsDefinition
# daily = DailyPartitionsDefinition(start_date="2024-01-01")
#
# @asset(partitions_def=daily)
# def raw_orders(context) -> pd.DataFrame:
#     day = context.partition_key
#     return fetch_orders(day)
#
# @asset(partitions_def=daily)
# def clean_orders(raw_orders: pd.DataFrame) -> pd.DataFrame:
#     return raw_orders.drop_duplicates("order_id")
#
# @asset(partitions_def=daily)
# def daily_revenue(clean_orders: pd.DataFrame) -> pd.DataFrame:
#     return clean_orders.groupby("region")["amount"].sum().reset_index()
#
# The dependency graph comes from the function SIGNATURES. You declare
# the tables that should exist; the scheduler works out the order and
# knows which downstream assets are stale when an upstream one changes.

# ---- WHAT AN ORCHESTRATOR GIVES YOU OVER CRON ----
CAPABILITIES = """
dependencies   B waits for A, and does not run if A failed
parallelism    independent branches run concurrently, automatically
retries        transient failures recover without human involvement
backfill       rerun a date range with the same code, no edits
observability  which task failed, its logs, its duration, its history
SLAs           alert when a task has not finished by an expected time
idempotency    a rerun of one task does not require rerunning everything
lineage        which outputs depend on which inputs
"""

# ---- SENSORS: event-driven rather than time-driven ----
# @task.sensor(poke_interval=300, timeout=7200, mode="reschedule")
def wait_for_upstream_file(ds):
    """Wait for a partner's file instead of guessing when it lands."""
    # mode="reschedule" frees the worker slot between checks; mode="poke"
    # holds it, and a few poke-mode sensors can starve the whole pool.
    return file_exists(f"s3://partner-drop/orders_{ds}.csv")

# ---- STRUCTURING TASKS: the boundaries that matter ----
GUIDELINES = """
1. One task = one unit of retry. If half a task succeeded and retrying
   redoes that half, the boundary is wrong.
2. Tasks must be idempotent. The orchestrator WILL rerun them.
3. Pass references (paths, ids), not data. Airflow XComs go through the
   metadata database; a DataFrame there will break it.
4. Parameterize by the logical date, never by now().
5. Put validation between extract and publish, so a bad run fails
   before anyone sees the numbers.
6. Set an execution timeout on everything. A hung task blocks the slot
   and often the next scheduled run.
"""

  A PIPELINE AS A DAG

        extract_orders ----\
                            +--> validate --> transform --> publish
        extract_refunds ---/                       |
                                                    \--> train_model

  the two extracts run in PARALLEL; validate waits for both
  publish and train_model both depend on transform, and run in parallel
  ACYCLIC: no task may depend, directly or indirectly, on itself

  LOGICAL DATE VERSUS RUN TIME

  logical date   2024-06-01     the day the data DESCRIBES
  actual run     2024-06-02 03:00   after the day has closed

  Code written against now() cannot be backfilled: rerunning it in
  August processes August. Code written against ds can rebuild any day.

  CRON                              ORCHESTRATOR
  0 2 * * * extract.sh              dependency-aware scheduling
  0 3 * * * transform.sh            retries with backoff
  0 4 * * * publish.sh              per-task logs and history
                                    backfill a date range
  transform runs at 03:00 even      parameterized by logical date
  if extract failed at 02:00        alerting and SLAs
  -> publishes yesterday's data
     as if it were today's

Key Takeaways

• An orchestrator models tasks and dependencies as a DAG, giving ordering, parallelism, retries, backfill, and per-task visibility that cron lacks.
• Parameterize every task by the logical execution date rather than now(), which is the property that makes backfills possible.
• One task should equal one unit of retry, and every task must be idempotent because the scheduler will rerun it.
• Pass paths and identifiers between tasks rather than data, since inter-task channels are metadata stores.
• Put validation between extraction and publication so a bad run fails before anyone reads the numbers.

🧪 Practice

1. Model a four-task pipeline as a DAG with two parallel extracts, and state which tasks rerun if the third task fails.
2. Rewrite a script that uses date.today() so it accepts a logical date, then backfill it over a week.
3. Interview: A cron-scheduled pipeline published yesterday's numbers as today's after an upstream outage. Which orchestrator feature would have prevented it? (Hint: what did the downstream job fail to check?)

Incremental Loading and Change Data Capture

Reloading an entire table every night works until the table is large, at which point it stops working abruptly rather than gradually. Incremental loading processes only what changed, and the whole difficulty is knowing what changed.

The simplest approach is a high-water mark: store the maximum updated_at from the last run and select rows newer than it. This is easy and has three specific failure modes that cause silent data loss.

First, clock skew and in-flight transactions. A row committed at 10:00:00 may become visible at 10:00:03, after a run that used 10:00:01 as its cutoff. That row is never picked up, by any future run, because the watermark has moved past it. The fix is a lookback overlap plus an idempotent merge.

Second, hard deletes are invisible. A row removed from the source simply stops appearing in queries, so the warehouse keeps it forever. Only soft deletes or change data capture solve this.

Third, the source must actually maintain the column. Many applications update rows without touching updated_at, and the field is then a partial record of changes rather than a complete one.

Change data capture reads the database's transaction log, the same stream the database uses for replication. Because every committed change is written there before anything else, CDC captures inserts, updates, and deletes with no reliance on application discipline and no query load on the source. It is the correct answer when you can have it.

The pattern that makes any of this safe is the merge, or upsert: match on a key, update when found, insert when not. It makes reprocessing the same records harmless, which is what lets you use overlapping windows.

import pandas as pd
from datetime import datetime, timedelta, timezone
from sqlalchemy import text

# ---- HIGH-WATER MARK, done safely ----
def incremental_load(source, warehouse, table, key="id",
                     ts_col="updated_at", overlap_minutes=15):
    """Load changed rows, with an overlap window and an idempotent merge."""
    last = warehouse.scalar(
        text("SELECT watermark FROM etl_state WHERE table_name = :t"),
        {"t": table},
    ) or datetime(1970, 1, 1, tzinfo=timezone.utc)

    # THE OVERLAP: go back further than strictly necessary. Combined with
    # a merge, reprocessing a few rows is free; missing one is permanent.
    since = last - timedelta(minutes=overlap_minutes)

    # Bound the top end too, so rows arriving mid-query are not half-read.
    until = datetime.now(timezone.utc) - timedelta(minutes=1)

    changed = source.read_sql(
        text(f"SELECT * FROM {table} "
             f"WHERE {ts_col} > :since AND {ts_col} <= :until"),
        params={"since": since, "until": until},
    )
    if changed.empty:
        return 0

    merge_upsert(warehouse, table, changed, key=key)

    # Advance the watermark from the DATA, not from the clock: if the
    # source lags, the clock would skip rows that had not yet arrived.
    new_watermark = min(changed[ts_col].max(), until)
    warehouse.execute(
        text("""INSERT INTO etl_state (table_name, watermark)
                VALUES (:t, :w)
                ON CONFLICT (table_name) DO UPDATE SET watermark = :w"""),
        {"t": table, "w": new_watermark},
    )
    return len(changed)

def merge_upsert(warehouse, table, df, key):
    """Load to a staging table, then MERGE. Idempotent by construction."""
    staging = f"staging_{table}"
    df.to_sql(staging, warehouse, if_exists="replace", index=False)
    cols = [c for c in df.columns if c != key]
    sets = ", ".join(f"{c} = EXCLUDED.{c}" for c in cols)
    warehouse.execute(text(f"""
        INSERT INTO {table} SELECT * FROM {staging}
        ON CONFLICT ({key}) DO UPDATE SET {sets}
    """))
# Re-inserting the same row updates it to the same values. That is what
# makes the overlap window free, and what makes a rerun after a crash
# safe rather than duplicating.

# ---- THE THREE FAILURE MODES, made concrete ----
FAILURES = """
1. IN-FLIGHT COMMIT
   10:00:00  transaction begins, sets updated_at = 10:00:00
   10:00:01  ETL runs with cutoff 10:00:01, does NOT see the row
   10:00:03  transaction commits; the row becomes visible
   10:00:01 is now the watermark, so the row is missed FOREVER.
   -> fix: overlap window + merge, or read the transaction log instead.

2. HARD DELETE
   The source runs DELETE FROM orders WHERE id = 42.
   No row appears in any incremental query, because there is no row.
   The warehouse keeps order 42 indefinitely.
   -> fix: soft deletes (deleted_at), periodic reconciliation, or CDC.

3. UNMAINTAINED TIMESTAMP
   An admin tool issues UPDATE orders SET status='x' WHERE id=7,
   without touching updated_at. The change is invisible to the loader.
   -> fix: a database trigger that maintains updated_at, or CDC.
"""

# ---- RECONCILIATION: catch what incremental loading missed ----
def reconcile(source, warehouse, table, key="id", sample_days=7):
    """Periodically compare source and target. Non-negotiable with
    watermark-based loading."""
    since = datetime.now(timezone.utc) - timedelta(days=sample_days)
    src = source.read_sql(
        text(f"SELECT {key}, updated_at FROM {table} WHERE updated_at > :s"),
        params={"s": since})
    tgt = warehouse.read_sql(
        text(f"SELECT {key}, updated_at FROM {table} WHERE updated_at > :s"),
        params={"s": since})

    merged = src.merge(tgt, on=key, how="outer", indicator=True,
                       suffixes=("_src", "_tgt"))
    missing = merged.loc[merged["_merge"] == "left_only", key]
    orphaned = merged.loc[merged["_merge"] == "right_only", key]
    stale = merged.loc[
        (merged["_merge"] == "both")
        & (merged["updated_at_src"] > merged["updated_at_tgt"]), key]

    report = {"missing_in_target": len(missing),
              "deleted_in_source": len(orphaned),
              "stale_in_target": len(stale)}
    if any(report.values()):
        print(f"RECONCILIATION DRIFT on {table}: {report}")
    return report
# Running this weekly turns silent divergence into a visible number.
# Without it, a warehouse can drift from its source for months.

# ---- CHANGE DATA CAPTURE: read the log, not the table ----
CDC_EVENT = {
    "op": "u",                                   # c create, u update, d delete
    "ts_ms": 1717236000000,
    "source": {"db": "shop", "table": "orders", "lsn": 24580923,
               "txId": 998877},
    "before": {"id": 42, "status": "pending", "amount": 50.0},
    "after":  {"id": 42, "status": "shipped", "amount": 50.0},
}

def apply_cdc(event, warehouse):
    op = event["op"]
    if op in ("c", "r"):                         # create, or initial snapshot
        upsert(warehouse, event["after"])
    elif op == "u":
        upsert(warehouse, event["after"])
    elif op == "d":
        # The DELETE is visible here, which is the whole point.
        soft_delete(warehouse, event["before"]["id"], event["ts_ms"])

# WHY CDC IS BETTER:
#   - captures deletes, which no query-based method can see
#   - no reliance on the application maintaining a timestamp
#   - no query load on the source database
#   - preserves the full ordered history of every change
#   - latency in seconds rather than one batch interval
# COST: operational complexity, replication slots that can fill the
# source's disk if a consumer stalls, and schema changes that must be
# handled in the stream.

# ---- APPEND-ONLY SOURCES: the easy case ----
def load_append_only(source, warehouse, table, id_col="event_id"):
    """Immutable events: the watermark can be the max id."""
    last_id = warehouse.scalar(text(f"SELECT COALESCE(MAX({id_col}), 0) "
                                    f"FROM {table}"))
    new = source.read_sql(text(f"SELECT * FROM {table} WHERE {id_col} > :i"),
                          params={"i": last_id})
    new.to_sql(table, warehouse, if_exists="append", index=False)
    return len(new)
# Safe only if ids are monotonically assigned AND committed in order.
# With gaps from rolled-back transactions this still works; with ids
# assigned before commit it has the same in-flight problem as timestamps.

Strategy	Catches deletes	Source load	Latency	Complexity
Full reload	yes	very high	one interval	lowest
Watermark on updated_at	no	moderate	one interval	low
Watermark plus overlap and merge	no	moderate	one interval	low
Append-only max id	n/a	low	one interval	low
CDC from the transaction log	yes	negligible	seconds	high

Key Takeaways

• Watermark loading misses rows committed after the cutoff was read, and those rows are lost permanently rather than picked up later.
• Combine an overlap window with an idempotent merge, which makes reprocessing free and missing a row much less likely.
• Query-based incremental loading cannot see hard deletes; only soft deletes, reconciliation, or CDC can.
• Advance the watermark from the data's maximum timestamp, not from the wall clock, so a lagging source does not cause skipped rows.
• CDC reads the transaction log, capturing every change including deletes without relying on application discipline.

🧪 Practice

1. Implement watermark loading with an overlap window and a merge, then demonstrate that running it twice over the same period changes nothing.
2. Write a reconciliation function reporting rows missing in the target, deleted in the source, and stale, and run it against seeded drift.
3. Interview: Your warehouse row count exceeds the source's and grows slowly each month. What is the most likely cause? (Hint: which operation leaves no trace in an updated_at column?)

Pipeline Idempotency and Retries

Pipelines fail. Networks drop, sources go down, a node is preempted, a deploy lands mid-run. The design question is not how to prevent failure but how to make recovery safe, and the property that makes it safe is idempotency.

An operation is idempotent if performing it multiple times has the same effect as performing it once. SET balance = 100 is idempotent; balance = balance + 10 is not. For a pipeline, idempotency means a rerun produces the same final state, which is what allows a retry to be automatic rather than a manual investigation.

This matters because every distributed system delivers at-least-once. A worker may complete its write and crash before recording success, so the orchestrator retries something that already happened. Without idempotency that retry duplicates data, and duplicated revenue is worse than missing revenue because it looks plausible.

Three patterns give idempotency. Delete-then-insert for a partition: the job removes the window it is about to write, so a rerun replaces rather than appends. Merge on a key: match and update, or insert. Deduplication on a stable event id: a unique constraint makes the second insert a no-op.

Retries need discipline too. Retry only transient failures, since retrying a schema error just fails five more times. Use exponential backoff with jitter so retries spread out instead of synchronising. Cap attempts and then fail loudly. And add a circuit breaker when the downstream is clearly down, so you stop hammering it.

The complementary practice is atomic publication: write to a temporary location, validate, then swap it into place in one operation. Readers see either the old complete version or the new complete version, never a partial one.

import time, random, hashlib, json
from datetime import datetime, timezone
from functools import wraps
import pandas as pd

# ---- NOT IDEMPOTENT: the default, and the bug ----
def append_daily(df, warehouse, table):
    df.to_sql(table, warehouse, if_exists="append", index=False)
    # Rerun after a partial failure and every row is duplicated. Totals
    # double. Nothing raises. A dashboard shows a great day.

# ---- PATTERN 1: DELETE-THEN-INSERT within a transaction ----
def load_partition(df, warehouse, table, partition_col, partition_value):
    """Replace one partition atomically. A rerun is a no-op in effect."""
    with warehouse.begin() as conn:               # all or nothing
        conn.execute(
            text(f"DELETE FROM {table} WHERE {partition_col} = :v"),
            {"v": partition_value},
        )
        df.to_sql(table, conn, if_exists="append", index=False)
# Without the transaction, a crash between DELETE and INSERT leaves the
# partition EMPTY, which is worse than either outcome.

# ---- PATTERN 2: MERGE on a business key ----
MERGE_SQL = """
INSERT INTO fact_orders (order_id, customer_id, amount, updated_at)
SELECT order_id, customer_id, amount, updated_at FROM staging_orders
ON CONFLICT (order_id) DO UPDATE SET
    customer_id = EXCLUDED.customer_id,
    amount      = EXCLUDED.amount,
    updated_at  = EXCLUDED.updated_at
WHERE fact_orders.updated_at < EXCLUDED.updated_at;   -- only if NEWER
"""
# The WHERE clause prevents an out-of-order replay from overwriting a
# newer record with an older one, which at-least-once delivery makes
# a real possibility rather than a theoretical one.

# ---- PATTERN 3: DEDUPLICATE on a stable event id ----
def stable_event_id(event: dict) -> str:
    """Derive a deterministic id when the source provides none."""
    canonical = json.dumps(
        {k: event[k] for k in sorted(event) if not k.startswith("_")},
        sort_keys=True, default=str,
    )
    return hashlib.sha256(canonical.encode()).hexdigest()[:32]
# It must be derived from the CONTENT. Using uuid4() or a timestamp
# produces a different id on each retry, defeating the deduplication.

e = {"order_id": "o_1", "amount": 59.9, "_received_at": "10:00"}
print(stable_event_id(e) == stable_event_id({**e, "_received_at": "10:05"}))
# True: the ingestion metadata is excluded, so a redelivery matches.

# ---- ATOMIC PUBLICATION: write aside, validate, then swap ----
def publish_atomically(df, warehouse, target_table, validators):
    temp = f"{target_table}__staging"
    df.to_sql(temp, warehouse, if_exists="replace", index=False)

    for name, check in validators.items():
        if not check(df):
            warehouse.execute(text(f"DROP TABLE {temp}"))
            raise ValueError(f"validation '{name}' failed; target untouched")

    with warehouse.begin() as conn:               # the swap is atomic
        conn.execute(text(f"ALTER TABLE {target_table} "
                          f"RENAME TO {target_table}__old"))
        conn.execute(text(f"ALTER TABLE {temp} RENAME TO {target_table}"))
        conn.execute(text(f"DROP TABLE {target_table}__old"))

validators = {
    "not_empty": lambda d: len(d) > 0,
    "no_null_keys": lambda d: d["order_id"].notna().all(),
    "unique_keys": lambda d: not d["order_id"].duplicated().any(),
    "plausible_volume": lambda d: 0.5 <= len(d) / 10_000 <= 2.0,
}
# The volume check catches the failure mode no schema test finds: a
# structurally valid table containing a tenth of the expected rows
# because an upstream filter changed.

# ---- RETRIES: only for transient failures ----
class TransientError(Exception): ...
class PermanentError(Exception): ...

def retry(attempts=5, base=1.0, cap=60.0, retry_on=(TransientError,)):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for i in range(attempts):
                try:
                    return fn(*args, **kwargs)
                except retry_on as exc:
                    if i == attempts - 1:
                        raise
                    delay = min(cap, base * 2 ** i) * (0.5 + random.random())
                    print(f"  attempt {i+1} failed ({exc}); retry in {delay:.1f}s")
                    time.sleep(delay)
                except PermanentError:
                    raise            # a schema error will never self-heal
        return wrapper
    return decorator

@retry(attempts=4, base=0.1)
def flaky_extract(state={"n": 0}):
    state["n"] += 1
    if state["n"] < 3:
        raise TransientError("connection reset")
    return "ok"

print(flaky_extract())

CLASSIFICATION = """
TRANSIENT, retry:        connection reset, timeout, 429, 503,
                         deadlock, node preempted, S3 slow-down
PERMANENT, do NOT retry: authentication failure, schema mismatch,
                         malformed input, 400, 403, disk full,
                         constraint violation
Retrying a permanent failure wastes the retry budget and delays the
alert that a human needs to see.
"""

# ---- CIRCUIT BREAKER: stop hammering a downstream that is down ----
class CircuitBreaker:
    def __init__(self, threshold=5, reset_after_s=60):
        self.threshold, self.reset_after = threshold, reset_after_s
        self.failures, self.opened_at = 0, None

    def call(self, fn, *a, **kw):
        if self.opened_at and time.time() - self.opened_at < self.reset_after:
            raise PermanentError("circuit open; downstream is failing")
        try:
            result = fn(*a, **kw)
            self.failures, self.opened_at = 0, None      # recovered
            return result
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.opened_at = time.time()
            raise

# ---- CHECKPOINTING long jobs so a retry resumes ----
def process_with_checkpoint(items, checkpoint_path, process_fn):
    from pathlib import Path
    done = set()
    p = Path(checkpoint_path)
    if p.exists():
        done = set(p.read_text().split())
    with open(p, "a") as fh:
        for item in items:
            if item in done:
                continue                    # already processed; skip
            process_fn(item)
            fh.write(f"{item}\n")
            fh.flush()                      # durable BEFORE continuing
# A three-hour job that fails at 90% resumes at 90% rather than zero.
# The flush matters: buffered writes lost in a crash cause reprocessing.

# ---- THE IDEMPOTENCY TEST, worth writing for every loader ----
def assert_idempotent(load_fn, warehouse, table, *args):
    load_fn(*args)
    first = warehouse.scalar(text(f"SELECT COUNT(*), SUM(amount) FROM {table}"))
    load_fn(*args)                          # run it again
    second = warehouse.scalar(text(f"SELECT COUNT(*), SUM(amount) FROM {table}"))
    assert first == second, f"NOT idempotent: {first} then {second}"

  WHY AT-LEAST-ONCE IS UNAVOIDABLE

  worker                          orchestrator
    | write 10,000 rows  -------> committed
    | record success ----X        (crash before the ack lands)
                                  no success recorded -> RETRY
    | write 10,000 rows  -------> committed AGAIN

  The orchestrator cannot distinguish "the write failed" from "the write
  succeeded and the acknowledgement was lost". It must assume the worst
  and retry. Therefore the WRITE must be idempotent.

  THE THREE PATTERNS

  delete-then-insert   DELETE WHERE day = '2024-06-01'; INSERT ...
                       in ONE transaction. Rerun replaces.

  merge on key         INSERT ... ON CONFLICT (id) DO UPDATE ...
                       Rerun updates to the same values.

  dedupe on event id   UNIQUE(event_id) + DO NOTHING.
                       The id must be derived from CONTENT, not
                       generated per attempt.

  ATOMIC PUBLICATION

  write temp -> validate -> rename into place
  readers see the OLD complete table or the NEW complete table,
  never a table mid-write

Key Takeaways

• Every distributed system delivers at least once, because a worker cannot distinguish a failed write from a lost acknowledgement.
• Idempotency comes from delete-then-insert within a transaction, merge on a key, or deduplication on a content-derived event id.
• Retry transient failures with exponential backoff and jitter, and fail immediately on permanent ones such as schema or authentication errors.
• Publish atomically by writing aside, validating, and swapping, so readers never see a partial table.
• A row-count plausibility check catches the failure that schema tests miss: a valid table with the wrong amount of data.

🧪 Practice

1. Write a loader using delete-then-insert in a transaction, then assert that running it twice leaves row count and column totals unchanged.
2. Implement a retry decorator distinguishing transient from permanent errors, and show it giving up immediately on a schema mismatch.
3. Interview: A pipeline retried once after a network blip and the daily revenue figure doubled. What property was missing and which of the three patterns would you apply? (Hint: what did the second run do that the first had already done?)

Data Contracts and Schema Evolution

The most common cause of a broken pipeline is not a bug in the pipeline. It is an upstream change nobody told you about: a column renamed, a type widened, an enum gaining a value, a field that was always populated becoming optional.

A data contract is an explicit agreement between a producer and a consumer about the shape and semantics of the data: field names, types, nullability, allowed values, expected volume, freshness, and who owns it. Making it explicit converts an implicit dependency into a reviewable artifact, so a breaking change fails in the producer's CI rather than in your dashboard.

The distinction that organises everything is backward versus forward compatibility. A change is backward compatible if new data can be read by old consumers, which is what you need when producers deploy first. It is forward compatible if old data can be read by new consumers, which you need when consumers deploy first. Adding an optional field with a default is both; removing a required field is neither.

The safe changes are short to list: add an optional field with a default, add a value to an enum that consumers treat permissively, widen a numeric type. The unsafe ones are equally short: remove or rename a field, narrow a type, make an optional field required, change a field's meaning while keeping its name. That last one is the worst, because nothing detects it.

Schema registries enforce this automatically. Producers register a schema, the registry checks compatibility against previous versions, and an incompatible change is rejected at publish time rather than discovered at read time.

The safe migration pattern for a breaking change is expand then contract: add the new field, dual-write both, migrate consumers, then remove the old field. It takes three deployments and no outages.

from dataclasses import dataclass, field
from typing import Optional, Literal
from datetime import datetime, timedelta
import pandas as pd

# ---- A DATA CONTRACT, expressed as code ----
from pydantic import BaseModel, Field, field_validator

class OrderEvent(BaseModel):
    """Contract v2. Owner: orders-team. SLA: 99.5% within 5 minutes."""
    model_config = {"extra": "forbid"}     # unknown fields are an ERROR

    order_id: str = Field(..., min_length=1, max_length=64)
    customer_id: str
    amount: float = Field(..., ge=0)
    currency: Literal["EUR", "USD", "GBP"]
    status: str
    occurred_at: datetime
    # v2 additions: OPTIONAL with defaults, so v1 producers stay valid.
    channel: Optional[str] = None
    discount: float = 0.0

    @field_validator("status")
    @classmethod
    def known_status(cls, v: str) -> str:
        known = {"pending", "shipped", "cancelled", "refunded"}
        if v not in known:
            # Log rather than raise: an unknown enum value should be
            # visible, not fatal, or every producer release breaks you.
            print(f"WARNING: unrecognised status '{v}'")
        return v

ok = OrderEvent(order_id="o_1", customer_id="c_1", amount=59.9,
                currency="EUR", status="shipped",
                occurred_at="2024-06-01T10:00:00Z")
print(ok.channel, ok.discount)             # None 0.0 -- defaults applied

try:
    OrderEvent(order_id="o_2", customer_id="c_2", amount=-5,
               currency="EUR", status="shipped",
               occurred_at="2024-06-01T10:00:00Z")
except Exception as exc:
    print("rejected:", str(exc).splitlines()[1].strip())

# extra="forbid" is a deliberate choice. It catches a producer adding a
# field you have not agreed to, which is often the first sign of a
# semantic change. The alternative, "ignore", is more permissive and
# lets silent changes through.

# ---- COMPATIBILITY, classified ----
COMPATIBILITY = """
                                   backward   forward   safe?
add optional field with default      yes        yes      YES
add required field                   no         yes      no
remove optional field                yes        no       careful
remove required field                no         no       NO
rename a field                       no         no       NO (= remove + add)
widen int32 -> int64                 yes        no       usually yes
narrow int64 -> int32                no         yes      NO
make optional -> required            no         yes      NO
add an enum value                    yes*       no       yes if consumers
                                                         handle unknowns
change MEANING, same name            yes        yes      WORST: nothing
                                                         detects it

backward = NEW data readable by OLD consumers  (producer deploys first)
forward  = OLD data readable by NEW consumers  (consumer deploys first)
FULL compatibility = both. Require it when deploy order is not controlled.
"""

# ---- EXPAND AND CONTRACT: renaming without an outage ----
MIGRATION = """
GOAL: rename `amount` to `gross_amount`.

Phase 1 EXPAND     add gross_amount, populate BOTH on every write.
                   Old consumers read amount. Nothing breaks.
                   ALTER TABLE orders ADD COLUMN gross_amount NUMERIC;
                   UPDATE orders SET gross_amount = amount;
                   -- plus application dual-write

Phase 2 MIGRATE    move consumers to gross_amount, one at a time.
                   Monitor reads of the old column to confirm none remain.

Phase 3 CONTRACT   stop writing amount, then drop it.
                   ALTER TABLE orders DROP COLUMN amount;

Three deploys, zero coordinated downtime. The alternative, renaming in
one step, requires every consumer to deploy in the same instant.
"""

# ---- DETECTING DRIFT before it reaches consumers ----
def schema_diff(expected: dict, actual: pd.DataFrame) -> dict:
    actual_types = {c: str(t) for c, t in actual.dtypes.items()}
    report = {
        "missing": sorted(set(expected) - set(actual_types)),
        "unexpected": sorted(set(actual_types) - set(expected)),
        "type_changed": {
            c: (expected[c], actual_types[c])
            for c in set(expected) & set(actual_types)
            if expected[c] != actual_types[c]
        },
    }
    report["breaking"] = bool(report["missing"] or report["type_changed"])
    return report

expected = {"order_id": "object", "amount": "float64", "status": "object"}
arrived = pd.DataFrame({"order_id": ["o_1"], "amount": ["59.90"],
                        "channel": ["web"]})
print(schema_diff(expected, arrived))
# missing: ['status']            a consumer will KeyError
# unexpected: ['channel']        probably harmless, worth noting
# type_changed: amount float->object   arithmetic now fails or concatenates

# ---- SEMANTIC DRIFT: the change no schema check finds ----
def semantic_monitor(df, column, baseline_stats, tolerance=0.3):
    """Types unchanged, meaning changed. Watch the distribution."""
    current = {"mean": df[column].mean(), "null_rate": df[column].isna().mean(),
               "distinct": df[column].nunique()}
    alerts = []
    for k, base in baseline_stats.items():
        cur = current[k]
        if base and abs(cur - base) / abs(base) > tolerance:
            alerts.append(f"{column}.{k}: {base:.4g} -> {cur:.4g}")
    return alerts

baseline = {"mean": 59.9, "null_rate": 0.01, "distinct": 5000}
today = pd.DataFrame({"amount": [5990.0] * 100})      # cents, not euros
print(semantic_monitor(today, "amount", baseline))
# The type is still float64 and every schema test passes. The upstream
# team switched the unit to cents. Only a distribution check catches it,
# and without one the error propagates into every downstream figure.

# ---- THE CONTRACT DOCUMENT ----
CONTRACT = """
dataset: orders.events
version: 2.1.0
owner: orders-team (slack #orders-eng, oncall: orders-oncall)
consumers:
  - analytics.fact_orders     (breaks if order_id or amount change)
  - ml.churn_features         (breaks if occurred_at changes)
schema: see OrderEvent (pydantic) / orders_v2.avsc (registry)
guarantees:
  grain:        one row per order state transition
  key:          (order_id, occurred_at) unique
  freshness:    p99 under 5 minutes
  completeness: order_id, amount, occurred_at never null
  volume:       8,000 to 20,000 events/day; outside this range, alert
  timezone:     occurred_at is UTC, always
  currency:     amount is in MAJOR units (euros, not cents)
change policy:
  additive changes:  announced in #data-contracts, no approval needed
  breaking changes:  30 days notice, expand-and-contract, consumer sign-off
deprecation:
  1.x supported until 2024-12-31
"""
# "amount is in MAJOR units" is the line that prevents the cents bug.
# Contracts must state SEMANTICS, not only types.

  WHERE THE BREAK HAPPENS, WITH AND WITHOUT A CONTRACT

  WITHOUT                           WITH
  producer deploys                  producer opens a PR
       |                                 |
  schema changes silently           CI checks the registry
       |                                 |
  pipeline runs, "succeeds"         INCOMPATIBLE -> build fails
       |                                 |
  dashboard shows wrong numbers     producer fixes it before merge
       |
  someone notices, days later
       |
  hours of investigation to find
  an upstream change

  The contract does not prevent change. It moves the failure from the
  consumer's dashboard to the producer's pull request.

  EXPAND AND CONTRACT

  t0   amount                 old consumers read amount
  t1   amount + gross_amount  both written; either works
  t2   amount + gross_amount  consumers migrated to gross_amount
  t3   gross_amount           amount dropped

  no instant at which a deployed consumer is broken

Key Takeaways

• A data contract makes an implicit dependency explicit, moving the failure from the consumer's dashboard to the producer's pull request.
• Backward compatibility lets new data reach old consumers; forward compatibility lets old data reach new consumers; require both when deploy order is not controlled.
• Adding an optional field with a default is safe, while renaming, removing, or narrowing is not.
• Use expand-and-contract to make breaking changes over three deploys with no coordinated downtime.
• Schema checks cannot detect a change of meaning under an unchanged type, so monitor distributions and state units in the contract.

🧪 Practice

1. Define a contract with a validation library, then feed it data with a missing field, a changed type, and an unexpected extra field.
2. Write a schema diff function classifying each difference as breaking or additive, and run it against two versions of a file.
3. Interview: An upstream team changes a monetary field from euros to cents without changing its type. Which of your checks catches it, and how quickly? (Hint: the type is identical, so what else changed?)

---

<a id="6-data-wrangling-and-preparation"></a>