A multi-file knowledge base for deep, production-grade data engineering. Each file goes far beyond surface treatment — internals, math, code samples, war stories, and the things that get asked in senior interview loops.

How to use this

Read in order if you're studying. Jump to a file if you're cramming for a specific area. The final file is interview Q&A built from real scenarios — page-at-3am incidents, design rounds, debugging walkthroughs — not leetcode.

Each file is self-contained. Code samples are runnable (Spark/PySpark, Flink, SQL on Snowflake/BigQuery/Postgres, Python). All examples assume Python 3.11+, Spark 3.5+, Flink 1.18+ unless stated otherwise.

Files

Reading paths

Cramming for an interview in a week: 08 first to know what you're aiming for, then 02/03 (processing), then 05 (SQL), then 01 (modeling), then 04 (Spark) and 07 (lakehouse) for depth on stack-specific questions.

Building a real system: 01 → 02 → 03 → 07 → 04. SQL and Python depth as needed.

Refresher / lookup: jump to whichever file has the topic. Each file's TOC is browsable.

What "deep" means here

Where the previous deep dive said "watermarks are a promise that no events with event_time ≤ W will arrive," this version derives the watermark formula, walks through how Flink computes it across operators with different latencies, shows the actual math for tuning bounded-out-of-orderness, and gives the code.

Where the previous version said "AQE handles skew automatically," this version shows the algorithm — how Spark detects a skewed partition, how it splits it, what the trade-offs are with broadcast vs shuffle, and the configs that govern each step.

That's the bar.

↑ Back to contents

The shape of your data at rest outlives every other choice. Code gets rewritten; schemas get migrated but never quite leave. This file goes deeper than "use a star schema" — it derives the decisions from first principles, shows the math that justifies them, and gives full DDL and MERGE patterns you can actually run.

1. Bounded vs Unbounded Data at Rest

Before storage models, the underlying question: is the dataset bounded (finite, the end is known) or unbounded (arriving forever)?

• Bounded at rest: a daily snapshot, a historical archive, a reference table. You can scan it, sort it, compute global aggregates, and re-derive it from source on demand.
• Unbounded at rest: an append-only log of events where you're modeling "everything that ever happened." Storage grows forever. Queries must always constrain time.

This distinction drives physical choices:

The engineering mistake is treating an unbounded stream-origin table as bounded — e.g., SELECT COUNT(DISTINCT user_id) FROM events without a dt filter. Over time this becomes unrunnable. Build guards into your table design (required partitions, partition-prune-or-fail settings like Snowflake's REQUIRE_PARTITION_FILTER, BigQuery's require_partition_filter=TRUE).

-- BigQuery: reject queries that don't filter on the partition column
CREATE TABLE playback.events_daily (
  event_ts      TIMESTAMP,
  user_id       STRING,
  title_id      STRING,
  watch_ms      INT64,
  dt            DATE
)
PARTITION BY dt
OPTIONS (
  require_partition_filter = TRUE,
  partition_expiration_days = 730   -- auto-drop after 2 years
);

-- Snowflake equivalent: rely on clustering + session param
ALTER SESSION SET QUERY_TAG = 'enforce_partition_filter';
-- Enforcement is done via resource monitors + query profiling, not DDL.

The unbounded nature also affects backfills. For a bounded dimension, "regenerate from source" is a few GB; for an unbounded fact, it's years of events.

2. OLTP vs OLAP — The Physical Reality

The choice between row-oriented and column-oriented storage isn't philosophical — it's dictated by what the hardware does with your queries.

Row-oriented storage (Postgres, MySQL, Oracle)

A row is laid out contiguously on disk. Reading row 42 = one random I/O + parse the row. Fast for "give me all columns of one row by primary key." Slow for "give me the average of column X across 100M rows" because every row must be loaded even though only one column is needed.

The storage page (typically 8 KB in Postgres, 16 KB in MySQL) is the unit of I/O. Postgres reads whole pages into shared_buffers. TID is (page_number, slot_number). B-tree indexes point at TIDs; looking up row 42 = B-tree walk (log N pages) + heap fetch.

Column-oriented storage (Parquet, ORC, Snowflake FDN, Redshift, BigQuery Capacitor)

Each column stored separately, with its own compression. Reading column X = read just X's bytes, skipping everything else. Compression ratios of 5–20× are routine because a single column has low cardinality locally.

A Parquet file is organized as:

• File footer: schema, row-group locations, column stats (min/max/null_count/distinct_count).
• Row groups (typically 128 MB): a horizontal slice of rows.
  • Column chunks: within a row group, one per column.
    • Pages (typically 1 MB): within a chunk, the unit of decoding.

The footer stats are the basis of row-group skipping — a WHERE filter can exclude row groups whose min/max don't match. This is why WHERE dt = '2026-04-16' AND country = 'US' is fast on a Parquet table partitioned by dt and sorted by country: both conditions prune.

Compression schemes within a column:

• RLE (Run-Length Encoding): AAAA BBBB CCCC → (A,4)(B,4)(C,4). Great on sorted columns with runs.
• Dictionary encoding: map values to ints; store ints + dictionary. Great on low-cardinality strings.
• Bit-packing: encode small integers in <8 bits each. Often combined with RLE.
• Delta encoding: store differences from previous value. Great on timestamps, sorted sequences.

The column-oriented nature has a direct modeling implication: adding a column is nearly free at read time (you only pay for it if you select it). Wide tables (OBT) aren't the disaster they'd be in a row store.

OLTP modeling implications

• Normalize: each update touches one row, so redundancy is expensive.
• B-tree indexes on everything queried. Covering indexes for hot queries.
• FK constraints enforced at write.
• Partitioning is a tool for manageability (vacuum, detach old partitions) more than performance.

OLAP modeling implications

• Denormalize within reason. Joins are expensive in distributed systems; repeated string values cost almost nothing with dictionary encoding.
• Sort / cluster by the most common filter key (usually date).
• No FK constraints; integrity is pipeline-enforced.
• Partitioning is the #1 physical layout decision. Get it wrong and queries scan entire years.

HTAP blur

New engines (TiDB, CockroachDB, SingleStore, Snowflake Hybrid Tables) offer row + column storage of the same logical table — writes go to the row store, reads to the column store with asynchronous replication. Nice for mid-size workloads; still limited in scale or consistency compared to pure OLTP/OLAP.

3. Normalization Derived From Functional Dependencies

Normalization is taught as a set of rules. It's actually a consequence of functional dependencies — statements of the form "column set X determines column set Y" (written X → Y).

Given dependencies:

• order_id → customer_id, order_date, status
• customer_id → customer_name, country

The relation Orders(order_id, customer_id, order_date, status, customer_name, country) violates 3NF because customer_id (non-key) determines customer_name, country — a transitive dependency. Decomposition:

• Orders(order_id, customer_id, order_date, status)
• Customers(customer_id, customer_name, country)

1NF — atomic values

No arrays, no comma-separated strings, no nested structures that the DB can't query. tags = "red,blue,green" forces LIKE '%blue%' queries that can't use indexes. Decompose into a child table order_tags(order_id, tag).

Modern engines allow arrays (Postgres text[], BigQuery ARRAY<STRING>) with GIN/ARRAY-function support. That's a departure from strict 1NF but acceptable when the array operations are well-supported.

2NF — full functional dependency on the key

Only relevant for composite keys. If PK is (order_id, line_number) and order_date depends on order_id alone, order_date doesn't belong in the line-item table. Move it.

3NF — no transitive dependencies

Shown above. In practice, 3NF is the ceiling for OLTP — further decomposition rarely helps.

BCNF (3.5NF) — every determinant is a candidate key

The textbook failure case: ClassRoom(course_id, instructor, room) where (course_id, instructor) → room and instructor → room. BCNF requires instructor to be a candidate key; it isn't, so decompose. These situations are rare in real data.

4NF & 5NF — multi-valued & join dependencies

Multi-valued dependencies arise when two independent multi-valued facts coexist: a teacher teaches multiple courses AND speaks multiple languages, and all combinations appear. 4NF says split these. 5NF handles even more arcane join dependencies. In 20 years of modeling work I've applied these consciously twice.

Practical heuristic

Normalize to 3NF. Denormalize deliberately, with a comment explaining why. A typical OLTP schema has 90% 3NF + 10% intentional denormalization (caching displayed values, storing denormalized totals for performance).

4. Dimensional Modeling — Kimball in Full

Kimball's dimensional model is a theory of analytic queries. It says: most analytics look like "a measure, broken down by attributes, filtered by other attributes." If your storage matches that shape, queries are trivial and fast.

The four-step process (and what each step actually means)

Step 1 — Business process. Not "the whole business" — a single process: orders, shipments, returns, playback sessions. Business processes are where events get generated; each is a candidate fact table. Get them from business users, not from source-system tables.

Step 2 — Grain. Declare exactly what one fact row represents. "One row per order line item per order" is a grain. "One row per playback session per user" is another. Grain must be atomic — prefer finer grain; you can always aggregate up.

The test: pick any two rows; they must represent genuinely distinct events. If two rows could be the same event counted twice, your grain is wrong.

Step 3 — Dimensions. The descriptors: who, what, where, when, why, how. For each dimension, identify the grain and attributes. Dimensions are denormalized (all attributes in one table) — you want category, subcategory, brand, department all present in dim_product even though they form a hierarchy.

Step 4 — Facts. The numeric measures at the declared grain. Prefer additive measures (quantity, extended_price). Document semi-additive (balance — additive across accounts but not time) and non-additive (ratio — don't sum ever).

Full star schema — playback analytics

-- -----------------------------------------------------
-- Date dimension (discussed more in §8)
-- -----------------------------------------------------
CREATE TABLE dim_date (
    date_key           INT PRIMARY KEY,             -- 20260419
    full_date          DATE NOT NULL,
    day_of_week        SMALLINT NOT NULL,           -- 1=Monday
    day_name           VARCHAR(10) NOT NULL,
    day_of_month       SMALLINT NOT NULL,
    day_of_year        SMALLINT NOT NULL,
    week_of_year       SMALLINT NOT NULL,           -- ISO week
    month_num          SMALLINT NOT NULL,
    month_name         VARCHAR(10) NOT NULL,
    quarter            SMALLINT NOT NULL,
    year               SMALLINT NOT NULL,
    fiscal_year        SMALLINT NOT NULL,
    fiscal_quarter     SMALLINT NOT NULL,
    fiscal_period      SMALLINT NOT NULL,
    is_weekend         BOOLEAN NOT NULL,
    is_holiday         BOOLEAN NOT NULL,
    holiday_name       VARCHAR(50),
    is_business_day    BOOLEAN NOT NULL,
    days_from_today    INT                          -- maintained via view
);

-- -----------------------------------------------------
-- User dimension — SCD Type 2
-- -----------------------------------------------------
CREATE TABLE dim_user (
    user_key           BIGINT PRIMARY KEY,          -- surrogate
    user_id            VARCHAR(36) NOT NULL,        -- natural/durable
    email              VARCHAR(320),
    country_code       CHAR(2),
    subscription_tier  VARCHAR(20),                 -- basic/standard/premium
    household_size     SMALLINT,
    profile_language   VARCHAR(10),
    signup_date        DATE,
    churn_date         DATE,
    valid_from         TIMESTAMPTZ NOT NULL,
    valid_to           TIMESTAMPTZ,
    is_current         BOOLEAN NOT NULL,
    hash_diff          CHAR(64) NOT NULL            -- SHA-256 of tracked attrs
);
CREATE UNIQUE INDEX dim_user_natural ON dim_user(user_id, valid_from);
CREATE INDEX dim_user_current ON dim_user(user_id) WHERE is_current;

-- -----------------------------------------------------
-- Title dimension — mostly Type 1, some Type 2 (rating)
-- -----------------------------------------------------
CREATE TABLE dim_title (
    title_key          BIGINT PRIMARY KEY,
    title_id           VARCHAR(36) NOT NULL,
    title              VARCHAR(500),
    title_type         VARCHAR(20),                 -- movie/series/documentary
    genre_primary      VARCHAR(50),
    genre_secondary    VARCHAR(50),
    runtime_minutes    INT,
    release_year       SMALLINT,
    language_primary   VARCHAR(10),
    content_rating     VARCHAR(20),                 -- MA, PG-13, etc.
    current_rating     NUMERIC(3,2),                -- aggregated, Type 1
    season_number      SMALLINT,
    episode_number     SMALLINT,
    is_original        BOOLEAN,
    valid_from         TIMESTAMPTZ NOT NULL,
    valid_to           TIMESTAMPTZ,
    is_current         BOOLEAN NOT NULL
);

-- -----------------------------------------------------
-- Device dimension — Type 1 (device facts don't "change" per user)
-- -----------------------------------------------------
CREATE TABLE dim_device (
    device_key         BIGINT PRIMARY KEY,
    device_id          VARCHAR(64) NOT NULL,
    device_type        VARCHAR(20),                 -- tv/mobile/tablet/web
    manufacturer       VARCHAR(50),
    model              VARCHAR(100),
    os                 VARCHAR(50),
    os_version         VARCHAR(20),
    app_version        VARCHAR(20)
);

-- -----------------------------------------------------
-- Playback session fact — transaction grain
-- One row per playback session (session starts and ends)
-- -----------------------------------------------------
CREATE TABLE fact_playback_session (
    session_key        BIGINT PRIMARY KEY,          -- surrogate per session
    session_id         VARCHAR(64) NOT NULL,        -- natural, from client
    user_key           BIGINT NOT NULL REFERENCES dim_user,
    title_key          BIGINT NOT NULL REFERENCES dim_title,
    device_key         BIGINT NOT NULL REFERENCES dim_device,
    start_date_key     INT NOT NULL REFERENCES dim_date,
    end_date_key       INT NOT NULL REFERENCES dim_date,
    start_ts           TIMESTAMPTZ NOT NULL,
    end_ts             TIMESTAMPTZ NOT NULL,
    -- Measures (all numeric, additive where possible)
    watch_ms           BIGINT NOT NULL,             -- additive
    buffer_ms          BIGINT NOT NULL,             -- additive
    seeks              INT NOT NULL,                -- additive
    pauses             INT NOT NULL,                -- additive
    max_bitrate_kbps   INT,                         -- non-additive (use avg)
    avg_bitrate_kbps   INT,                         -- non-additive (use avg)
    completion_ratio   NUMERIC(4,3),                -- non-additive ratio
    -- Degenerate dimensions
    ended_reason       VARCHAR(20),                 -- user/credits/error
    qoe_score          NUMERIC(3,2),
    -- Partition
    dt                 DATE NOT NULL
)
PARTITION BY RANGE (dt);

Key details people miss:

• Every FK is a surrogate _key, not a natural _id. The natural IDs are attributes, preserved for lookup.
• start_date_key and end_date_key both reference dim_date — same table, role-playing.
• session_id is stored as a degenerate dimension (an attribute on the fact with no corresponding dim table) because there's no other data to put in a dim_session.
• max_bitrate_kbps and avg_bitrate_kbps are non-additive measures. You need both if you want to compute weighted averages at higher grains.

The query this shape makes easy

-- Total watch hours by country and content rating, last 7 days, weekdays only
SELECT
    u.country_code,
    t.content_rating,
    SUM(f.watch_ms) / 3600.0 / 1000 AS watch_hours
FROM fact_playback_session f
JOIN dim_user  u ON u.user_key  = f.user_key
JOIN dim_title t ON t.title_key = f.title_key
JOIN dim_date  d ON d.date_key  = f.start_date_key
WHERE d.full_date >= CURRENT_DATE - 7
  AND d.is_weekend = FALSE
GROUP BY u.country_code, t.content_rating
ORDER BY watch_hours DESC;

No CASE statements, no date arithmetic, no EXTRACT — just filter and group. This is why Kimball wins.

Why surrogate keys

Four reasons, in increasing order of importance:

1. Natural keys change. Source systems renumber, reassign IDs during migrations. Your warehouse shouldn't break.
2. SCD2 requires multiple rows per natural ID — the surrogate disambiguates.
3. Integer surrogates are small and fast in joins. A BIGINT is 8 bytes; a VARCHAR(36) UUID is 36 bytes + overhead and comparison is slower.
4. Surrogates decouple the warehouse's keyspace from source systems. You can integrate two sources whose customer_id namespaces collide.

Generate surrogates deterministically with SHA-256 of the natural key (for idempotent re-runs) or from a sequence (for insertion-order keys). The deterministic approach is safer in modern pipelines.

-- Deterministic surrogate from natural key
SELECT
    HASH(user_id, valid_from) AS user_key,   -- Snowflake HASH(x1,...) -> BIGINT
    user_id, valid_from, ...
FROM staging_users;

-- Spark / PySpark
from pyspark.sql.functions import sha2, concat_ws, col
df = df.withColumn(
    "user_key",
    sha2(concat_ws("||", col("user_id"), col("valid_from")), 256)
)

Use a hashing function that's stable across restarts; Spark's built-in hash() isn't guaranteed stable across versions, but sha2 and md5 are.

5. Fact Table Grain (The Most Important Decision)

Grain is where every modeling interview should start. The failure mode is a fact table where some rows are at one grain and others at another. The bug is invisible until a sum is 2× too large.

Three transactional fact patterns

Transaction fact — one row per event. Most common. Fully additive. Example: fact_playback_session above.

Periodic snapshot fact — one row per entity per period. Measures describe the state at period end. Semi-additive.

CREATE TABLE fact_subscription_daily (
    snapshot_date_key  INT,
    user_key           BIGINT,
    -- Semi-additive: summable across users, NOT across days
    mrr_cents          BIGINT,       -- monthly recurring revenue
    status             VARCHAR(20),
    days_until_renewal SMALLINT,
    -- Additive within the day
    payments_made      INT,
    refunds_issued     INT,
    PRIMARY KEY (snapshot_date_key, user_key)
);

The MRR column is the classic semi-additive trap. Users total mrr across two days and get 2× the real revenue. Solutions: label the column in the metadata catalog; always aggregate via AVG across time and SUM across entity; make the column name end in _point_in_time.

Accumulating snapshot fact — one row per process instance, updated as it progresses through milestones.

CREATE TABLE fact_order_fulfillment (
    order_key         BIGINT PRIMARY KEY,
    customer_key      BIGINT,
    -- Milestones (date keys; NULL until reached)
    order_date_key    INT NOT NULL,
    paid_date_key     INT,
    packed_date_key   INT,
    shipped_date_key  INT,
    delivered_date_key INT,
    returned_date_key INT,
    -- Durations, derived at each update
    hours_to_pay      INT,
    hours_to_pack     INT,
    hours_to_ship     INT,
    hours_to_deliver  INT,
    -- Measures
    order_total_cents BIGINT,
    items_count       INT
);

Pros: a single row per order gives "lead time analysis" trivially. Cons: heavy on updates (updates are the enemy of columnar engines). Use when the process has bounded, well-defined milestones.

Additivity

Additivity is the single most-often-wrong attribute in a fact table. Get it wrong and every dashboard off that table is silently lying. The taxonomy is three-way, but the real skill is recognizing which bucket any given measure falls into across domains. The table below expands the usual starter examples into a cross-industry reference you can pattern-match against.

Domain-specific additivity cheat sheet

Use this to pattern-match new domains against known ones. The same measure shape shows up again and again.

Measures that look additive but aren't

The following look like counts — they are not safely sum-able without context. Each has a specific remediation.

The canonical non-additive mistake: you have fact_product_daily(product_key, date_key, avg_price). A BI user computes "average price this week = AVG(avg_price) across 7 days." That's an average of averages — not the weighted average they want. Fix: store sum_of_prices and price_count, compute the ratio in the SELECT. Same rule for everything in the third table above — the shape is always "store the components, derive the ratio on read."

Factless fact tables

A fact table with only FKs — no measures. Models events or eligibility:

-- "Which users had access to which titles on which day"
CREATE TABLE fact_title_availability_daily (
    date_key  INT,
    user_key  BIGINT,
    title_key BIGINT,
    -- no measures
    PRIMARY KEY (date_key, user_key, title_key)
);

The "measure" is COUNT(*) — "how many users had access to Stranger Things in France on 2026-04-01."

These get huge fast (N_users × N_titles × N_days). Sparse-encode where possible (don't emit rows for absent combinations).

Multi-grain fact: how to avoid it

The canonical multi-grain bug: an orders fact table where some rows are "one per order" and some are "one per line item." Summing order_total sums header rows plus line rows — each order's total appears many times.

Rules:

1. One fact table per declared grain.
2. If you need both grains, build two fact tables: fact_order_header and fact_order_line. They roll up via shared dimensions.
3. Never introduce an "aggregation type" column to disambiguate rows of different grains. That's a band-aid on a severed arm.

6. Dimensions — Conformed, Junk, Degenerate, Mini, Role-Playing

Conformed dimensions

A dimension is conformed when multiple fact tables share it. This is how you answer "return rate by customer segment" — fact_returns.customer_key and fact_orders.customer_key must reference the same dim_customer.

Conformed dimensions are an organizational commitment as much as a technical one. A dim_customer owned by marketing with fields marketing cares about, and another owned by billing with billing fields, is a failure. You need a single dim_customer with all stakeholders as consumers.

The enterprise bus matrix (Kimball's term) is a table of business processes × dimensions, with an X where they intersect. Use it to plan which dimensions must be conformed:

                      dim_user  dim_title  dim_device  dim_date  dim_geo  dim_promo
playback_session         X         X          X          X         X        -
billing_charge           X         -          -          X         X        X
search_query             X         -          X          X         X        -
signup                   X         -          X          X         X        X

Junk dimensions

When you have a handful of low-cardinality flags and indicators that don't belong together but don't warrant their own dimensions: combine them into a junk dimension.

-- Instead of: is_trial BOOLEAN, is_promo BOOLEAN, device_is_new BOOLEAN on every fact row...
CREATE TABLE dim_session_flags (
    flags_key   INT PRIMARY KEY,
    is_trial    BOOLEAN,
    is_promo    BOOLEAN,
    is_new_device BOOLEAN,
    signal_source VARCHAR(20)
);
-- Only 2x2x2x (distinct sources) = ~16 rows.

Then the fact has one flags_key FK and you get clean reporting.

Degenerate dimensions

An attribute on the fact with no corresponding dim table — like session_id above. Reason: there's no other data about a session beyond what's already in the fact row. Creating dim_session would just be a table with one column.

Mini-dimensions

When a dimension has a subset of frequently-changing attributes that would inflate SCD2 storage: split them off.

-- dim_user (stable attrs)
user_key, user_id, email, country_code, signup_date, ...

-- dim_user_demographics (mini-dim; snapshots of volatile attrs)
demographics_key, age_bracket, income_bracket, household_size, interest_segment, snapshot_ts

The fact has FKs to both: user_key (as-current) and demographics_key (as-at-event-time). Saves SCD2 storage when the volatile attributes change often and the stable ones don't.

Role-playing dimensions

One physical dimension, multiple roles in the fact. dim_date plays order_date, ship_date, delivered_date. Implement as views aliasing the base dim so query joins read naturally:

CREATE VIEW dim_order_date    AS SELECT * FROM dim_date;
CREATE VIEW dim_ship_date     AS SELECT * FROM dim_date;
CREATE VIEW dim_delivered_date AS SELECT * FROM dim_date;

SELECT
    od.year, od.month_num,
    sd.day_name AS ship_day,
    dd.week_of_year AS delivered_week,
    SUM(f.order_total_cents)
FROM fact_order_fulfillment f
JOIN dim_order_date      od ON od.date_key = f.order_date_key
JOIN dim_ship_date       sd ON sd.date_key = f.shipped_date_key
JOIN dim_delivered_date  dd ON dd.date_key = f.delivered_date_key
GROUP BY od.year, od.month_num, sd.day_name, dd.week_of_year;

The physical storage is one table; the logical model has three.

Swappable / outrigger

When a dimension attribute has its own attributes worth rolling up: a small normalized "outrigger" joined to the main dim. dim_product.category_id → dim_category.name, parent_category. Violates pure star; sometimes necessary.

7. Slowly Changing Dimensions — All Seven Types, with Code

Type 0 — Retain original

Never update. Fields like date_of_birth, signup_date, birth_country. No code needed beyond rejecting updates.

Type 1 — Overwrite

Just update. No history. Use for typo corrections, GDPR erasure of PII.

-- Stage has latest values; merge overwrites
MERGE INTO dim_user AS tgt
USING stg_user AS src
  ON tgt.user_id = src.user_id
WHEN MATCHED THEN UPDATE SET
    email         = src.email,
    country_code  = src.country_code,
    updated_at    = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN INSERT (...) VALUES (...);

Type 2 — Add new row, track history

The standard. Full history. Every change creates a new row; old row closed with valid_to.

The MERGE pattern (dialect-independent)

Implemented as two statements because a single MERGE can't both update and insert the same logical record.

-- Step 1: close out rows whose tracked attributes have changed
UPDATE dim_user
SET valid_to   = CURRENT_TIMESTAMP,
    is_current = FALSE
WHERE is_current = TRUE
  AND user_id IN (SELECT user_id FROM stg_user)
  AND (
    -- Use a hash_diff for concise comparison
    hash_diff <> (
      SELECT SHA2(CONCAT_WS('||',
        COALESCE(s.email, ''),
        COALESCE(s.country_code, ''),
        COALESCE(s.subscription_tier, ''),
        COALESCE(s.household_size::VARCHAR, ''),
        COALESCE(s.profile_language, '')
      ), 256)
      FROM stg_user s WHERE s.user_id = dim_user.user_id
    )
  );

-- Step 2: insert a row for every natural key with no current row
INSERT INTO dim_user (
    user_key, user_id, email, country_code, subscription_tier,
    household_size, profile_language, signup_date, churn_date,
    valid_from, valid_to, is_current, hash_diff
)
SELECT
    nextval('dim_user_seq'),                   -- surrogate
    s.user_id,
    s.email, s.country_code, s.subscription_tier,
    s.household_size, s.profile_language,
    s.signup_date, s.churn_date,
    CURRENT_TIMESTAMP, NULL, TRUE,
    SHA2(CONCAT_WS('||',
      COALESCE(s.email, ''),
      COALESCE(s.country_code, ''),
      COALESCE(s.subscription_tier, ''),
      COALESCE(s.household_size::VARCHAR, ''),
      COALESCE(s.profile_language, '')
    ), 256)
FROM stg_user s
LEFT JOIN dim_user d
  ON d.user_id = s.user_id AND d.is_current = TRUE
WHERE d.user_id IS NULL;

Why hash_diff: comparing every column individually is verbose, error-prone, and slow. Hashing the concatenated tracked columns lets you compare two states with a single =. Use SHA-256 (not MD5; collision-free enough for dim-change detection).

Why two statements: MERGE's WHEN MATCHED runs once per matched row. You can't both close the old row and insert a new one in the same MERGE without duplication. Separate steps are clearer anyway.

PySpark / Delta Lake version

from delta.tables import DeltaTable
from pyspark.sql.functions import sha2, concat_ws, coalesce, lit, col, current_timestamp, monotonically_increasing_id

dim = DeltaTable.forName(spark, "warehouse.dim_user")
stg = spark.read.table("stg_user")

# Add hash_diff to the stage
stg_h = stg.withColumn(
    "hash_diff",
    sha2(concat_ws("||",
        coalesce(col("email"), lit("")),
        coalesce(col("country_code"), lit("")),
        coalesce(col("subscription_tier"), lit("")),
        coalesce(col("household_size").cast("string"), lit("")),
        coalesce(col("profile_language"), lit(""))
    ), 256)
)

# Step 1: close rows whose hash changed
(dim.alias("t")
    .merge(stg_h.alias("s"),
           "t.user_id = s.user_id AND t.is_current = TRUE AND t.hash_diff <> s.hash_diff")
    .whenMatchedUpdate(set={
        "valid_to":   "current_timestamp()",
        "is_current": "FALSE"
    })
    .execute())

# Step 2: insert new versions
changed = (stg_h.alias("s")
    .join(dim.toDF().alias("t"),
          (col("s.user_id") == col("t.user_id")) & (col("t.is_current") == True),
          "left_anti"))   # keep s rows not present as current

changed_with_keys = changed.withColumn("user_key", monotonically_increasing_id() + <max_key>)
changed_with_keys.write.format("delta").mode("append").saveAsTable("warehouse.dim_user")

monotonically_increasing_id() per partition + offset is one pattern for surrogate generation. Deterministic hashing is safer across re-runs.

dbt snapshot (the declarative way)

# snapshots/dim_user.sql
{% snapshot dim_user_snapshot %}
{{ config(
    target_schema='snapshots',
    unique_key='user_id',
    strategy='check',
    check_cols=['email', 'country_code', 'subscription_tier',
                'household_size', 'profile_language']
) }}
SELECT user_id, email, country_code, subscription_tier,
       household_size, profile_language, signup_date, churn_date
FROM {{ source('raw', 'users') }}
{% endsnapshot %}

dbt handles the entire Type 2 lifecycle — adds dbt_valid_from, dbt_valid_to, compares columns, inserts new versions. Production-grade with one file.

Type 3 — Add new column (previous + current)

Tracks exactly one prior value. Niche — useful only when business cares about "previous region" specifically.

ALTER TABLE dim_user
  ADD COLUMN previous_country_code CHAR(2),
  ADD COLUMN country_changed_at TIMESTAMPTZ;

On update: UPDATE ... SET previous_country_code = country_code, country_code = new, country_changed_at = now().

Type 4 — History in a side table

Current row in the main dim (Type 1 semantics for ease of query); history in a separate table.

-- Main dim (current-state only)
CREATE TABLE dim_user_current (
    user_key BIGINT PRIMARY KEY,
    user_id VARCHAR(36) UNIQUE,
    email, country_code, ...
);

-- History (append-only, every change)
CREATE TABLE dim_user_history (
    user_id VARCHAR(36),
    email, country_code, ...,
    valid_from TIMESTAMPTZ,
    valid_to TIMESTAMPTZ,
    change_reason VARCHAR(50)
);

Good when the main dim is hit constantly by dashboards and you want a narrow table. Point-in-time joins use the history table.

Type 6 — 1 + 2 + 3

A Type 2 dim where every historical row also carries the current value of select attributes. Lets you report "sales by the customer's current segment" even on old facts.

CREATE TABLE dim_user (
    user_key BIGINT PRIMARY KEY,
    user_id  VARCHAR(36) NOT NULL,
    -- Historical values (Type 2 — different per version)
    email, country_code, subscription_tier,
    -- Current values (Type 1 — same across all versions of same user_id)
    current_email, current_country_code, current_subscription_tier,
    -- History metadata
    valid_from, valid_to, is_current,
    hash_diff
);

The Type 1 columns are updated across all rows with the same user_id whenever the current state changes — an expensive broadcast update, but query-time simplicity worth it.

Query patterns:

• "Watch hours by historical segment": join on user_key, group by subscription_tier.
• "Watch hours by current segment": join on user_key, group by current_subscription_tier.

Type 7 — Dual keys

The fact carries both the surrogate (user_key — version at event time) and the durable natural key (user_id). Consumers pick which to join on. Rare but powerful for complex reporting platforms.

ALTER TABLE fact_playback_session ADD COLUMN user_id VARCHAR(36);  -- durable

-- Historical view
JOIN dim_user d ON d.user_key = f.user_key

-- Current view (resolve through natural key)
JOIN dim_user_current d ON d.user_id = f.user_id

Picking a type per attribute

Attributes within one dimension almost always use different types. The skill is choosing correctly per attribute. The expanded table below covers the attributes you'll see in most real dimensions, with the reasoning.

Document every per-attribute decision in the data catalog. A newcomer should be able to read the catalog and predict whether an update to country_code produces a new row in dim_user. If the answer is ambiguous, the catalog is incomplete.

The red-flag shapes

Three attribute categories warrant extra suspicion when choosing an SCD type:

• Anything regulatory. Legal name, address, tax ID, consent flags — treat as Type 2 minimum, plus a separate audit trail, regardless of analytical need. Compliance trumps modeling elegance.
• Anything that drives financial attribution. Tier, plan, price, territory assignment, account executive ownership — always Type 2. A future auditor will ask "who owned this account when the deal closed." Type 1 cannot answer that.
• Anything computed. Lifetime value, segment classification, engagement score — don't SCD these; move them to a periodic snapshot fact. Dimensions should hold identity and attributes; computed aggregates belong with the facts.

8. Date & Time Dimensions Done Right

A first-class date dimension makes fiscal/holiday/weekend/quarter queries trivial. Generate it once, populate 20 years of rows, never touch again.

Generating dim_date (Postgres)

INSERT INTO dim_date (
    date_key, full_date, day_of_week, day_name,
    day_of_month, day_of_year, week_of_year,
    month_num, month_name, quarter, year,
    fiscal_year, fiscal_quarter, fiscal_period,
    is_weekend, is_holiday, holiday_name, is_business_day
)
SELECT
    TO_CHAR(d, 'YYYYMMDD')::INT                       AS date_key,
    d                                                 AS full_date,
    EXTRACT(ISODOW FROM d)                            AS day_of_week,
    TO_CHAR(d, 'FMDay')                               AS day_name,
    EXTRACT(DAY FROM d)                               AS day_of_month,
    EXTRACT(DOY FROM d)                               AS day_of_year,
    EXTRACT(WEEK FROM d)                              AS week_of_year,
    EXTRACT(MONTH FROM d)                             AS month_num,
    TO_CHAR(d, 'FMMonth')                             AS month_name,
    EXTRACT(QUARTER FROM d)                           AS quarter,
    EXTRACT(YEAR FROM d)                              AS year,
    -- Netflix fiscal year happens to be calendar; substitute your rules
    EXTRACT(YEAR FROM d)                              AS fiscal_year,
    EXTRACT(QUARTER FROM d)                           AS fiscal_quarter,
    EXTRACT(MONTH FROM d)                             AS fiscal_period,
    (EXTRACT(ISODOW FROM d) IN (6,7))                 AS is_weekend,
    FALSE                                             AS is_holiday,
    NULL                                              AS holiday_name,
    (EXTRACT(ISODOW FROM d) BETWEEN 1 AND 5)          AS is_business_day
FROM generate_series(DATE '2010-01-01', DATE '2040-12-31', INTERVAL '1 day') d;

-- Holidays (join a separate table, mark is_holiday)
UPDATE dim_date d SET is_holiday = TRUE, holiday_name = h.name, is_business_day = FALSE
FROM holidays h WHERE h.observed_date = d.full_date;

Time-of-day dimension (separate)

Never combine date + time into one dimension — cardinality explodes (30 years × 86400 seconds = ~1 billion rows).

CREATE TABLE dim_time (
    time_key     SMALLINT PRIMARY KEY,         -- 0..1439 (minute-of-day)
    hour         SMALLINT,
    minute       SMALLINT,
    period       VARCHAR(2),                   -- AM/PM
    hour_12      SMALLINT,
    time_of_day_segment VARCHAR(20),           -- 'morning', 'evening'
    is_peak_hours BOOLEAN
);
-- 1440 rows total.

Facts reference both: start_date_key + start_time_key from start_ts.

Timezone handling — the ironclad rule

Store everything in UTC. Convert for display at presentation.

• Source systems emit timestamps; normalize to UTC in the first transform (bronze → silver).
• Use TIMESTAMP WITH TIME ZONE columns (TIMESTAMPTZ) in Postgres/Redshift; TIMESTAMP in Snowflake/BigQuery is naive — wrap consistently.
• For local-time analytics ("plays during 8pm local per user"), precompute the local date/time on ingest using the user's known timezone. This avoids runtime AT TIME ZONE for millions of rows.

-- Compute user-local ts at ingest
SELECT
    event_ts AT TIME ZONE 'UTC' AT TIME ZONE u.timezone AS local_ts,
    (event_ts AT TIME ZONE 'UTC' AT TIME ZONE u.timezone)::DATE AS local_date
FROM raw_events e JOIN dim_user u ON u.user_id = e.user_id;

DST is the silent bug. Whenever you convert, be prepared for 2026-03-08 02:30 US/Eastern — which doesn't exist. Use library functions (pytz, Java ZonedDateTime) that raise errors rather than silently shift.

9. Data Vault 2.0 — When and Why

Data Vault is a modeling methodology for enterprises with many source systems, strong auditability requirements, and the need to evolve quickly without reshaping downstream.

Core structures

Hub — unique list of business keys.

CREATE TABLE hub_customer (
    customer_hk    CHAR(64) PRIMARY KEY,       -- hash of business key
    customer_bk    VARCHAR(100) NOT NULL,       -- natural / business key
    load_dts       TIMESTAMPTZ NOT NULL,
    record_source  VARCHAR(100) NOT NULL
);

Link — relationships between hubs.

CREATE TABLE link_order (
    order_hk       CHAR(64) PRIMARY KEY,        -- hash(customer_bk, product_bk, order_bk)
    customer_hk    CHAR(64) NOT NULL REFERENCES hub_customer,
    product_hk     CHAR(64) NOT NULL REFERENCES hub_product,
    order_bk       VARCHAR(100) NOT NULL,
    load_dts       TIMESTAMPTZ NOT NULL,
    record_source  VARCHAR(100) NOT NULL
);

Satellite — descriptive attributes, with full history built in.

CREATE TABLE sat_customer_details (
    customer_hk    CHAR(64) NOT NULL,
    load_dts       TIMESTAMPTZ NOT NULL,
    load_end_dts   TIMESTAMPTZ,
    hash_diff      CHAR(64) NOT NULL,
    -- Attributes
    email          VARCHAR(320),
    country_code   CHAR(2),
    subscription_tier VARCHAR(20),
    record_source  VARCHAR(100) NOT NULL,
    PRIMARY KEY (customer_hk, load_dts)
);

Loading pattern

Highly parallel — each hub, link, and sat can be loaded independently as long as hubs are loaded before their sats. No dependencies between sources.

-- Load pattern (per source, per hub)
INSERT INTO hub_customer (customer_hk, customer_bk, load_dts, record_source)
SELECT DISTINCT
    SHA2(s.customer_id, 256) AS customer_hk,
    s.customer_id,
    CURRENT_TIMESTAMP,
    'billing_system'
FROM stage_billing s
LEFT JOIN hub_customer h ON h.customer_hk = SHA2(s.customer_id, 256)
WHERE h.customer_hk IS NULL;

Pros

• Source-system-agnostic: the vault survives upstream changes with zero refactor.
• Auditable: every fact has load_dts + record_source; lineage is built in.
• Parallel loading: no cross-source dependencies.
• Adding new sources is an additive operation.

Cons

• Unqueryable directly. You build a Kimball-style "information mart" on top for BI. Two layers = more moving parts.
• Many joins: hubs + sats + links to get a single business view.
• Overkill for small/single-source warehouses.
• Team needs training; the discipline is easy to violate.

When to use it

• 50+ source systems to integrate.
• Regulatory compliance (insurance, banking, healthcare) requires full audit trail.
• Large data engineering team that can maintain both vault + marts.

When NOT to use it: small warehouses (<10 sources), teams without DV expertise, or when business moves too fast for a two-layer architecture.

10. One Big Table & the Columnar Revolution

Columnar compression changes the math. With dictionary encoding, 100M rows of "premium" cost roughly 100M × 1-byte-dictionary-code + one entry "premium" in the dictionary. ~100MB uncompressed. With RLE (runs of "premium"): a few bytes per run.

This means denormalizing a dimension into a fact table adds almost no storage. The join-elimination benefit is real. Hence OBT.

What OBT looks like

-- A wide, denormalized table: 200 columns, billions of rows, no joins required
CREATE TABLE gold.playback_sessions_enriched (
    -- Primary identifiers
    session_id           VARCHAR(64),
    session_key          BIGINT,
    -- User attributes (denormalized from dim_user)
    user_id              VARCHAR(36),
    user_country_code    CHAR(2),
    user_subscription_tier VARCHAR(20),
    user_household_size  SMALLINT,
    user_profile_language VARCHAR(10),
    user_signup_date     DATE,
    user_segment         VARCHAR(50),
    -- Title attributes
    title_id             VARCHAR(36),
    title                VARCHAR(500),
    title_type           VARCHAR(20),
    title_genre_primary  VARCHAR(50),
    title_runtime_min    INT,
    title_is_original    BOOLEAN,
    -- Device attributes
    device_type          VARCHAR(20),
    device_os            VARCHAR(50),
    -- Session measures
    watch_ms             BIGINT,
    buffer_ms            BIGINT,
    seeks                INT,
    pauses               INT,
    max_bitrate_kbps     INT,
    completion_ratio     NUMERIC(4,3),
    -- Derived (not in star)
    watched_in_peak_hours BOOLEAN,
    was_binge_session    BOOLEAN,
    qoe_score            NUMERIC(3,2),
    -- Time
    start_ts             TIMESTAMPTZ,
    end_ts               TIMESTAMPTZ,
    dt                   DATE
)
PARTITIONED BY (dt)
CLUSTER BY (user_country_code, title_genre_primary);

Consumer queries have zero joins:

SELECT
    user_country_code, title_genre_primary,
    SUM(watch_ms) / 3600e3 AS watch_hours
FROM gold.playback_sessions_enriched
WHERE dt BETWEEN '2026-04-12' AND '2026-04-18'
GROUP BY user_country_code, title_genre_primary;

Zone maps + clustering make this scan a tiny fraction of the table.

OBT trade-offs

Pros:

• One join-less table.
• Dashboards fly.
• Simple mental model for analysts.
• Compressed storage cost is often lower than the original star due to better dictionary encoding across a wider table.

Cons:

• SCD2 becomes painful. Do you carry every historical attribute on every fact row? Or pin-time user attributes as-of event time and skip "current" reporting? Either is a choice.
• Backfilling dimension changes is expensive. If user_segment logic changes, every historical row must be rewritten.
• Data contracts grow. 200 columns × many downstream consumers = a lot of coordination on changes.
• Governance nightmare. Sensitive fields proliferate — you now have email on billions of fact rows; GDPR deletes touch all of them.

Hybrid: Kimball silver + OBT gold

The pragmatic pattern most serious teams end up with:

• Silver = conformed star schema (dims + facts), Kimball-style.
• Gold = one OBT per consumption pattern, materialized from silver.

Regenerate gold cheaply when dimensions change. Silver provides the single source of truth. BI reads gold for speed.

11. Medallion Architecture (Bronze/Silver/Gold)

The lakehouse convention. Each layer has a distinct purpose:

Bronze — raw, append-only, preserved

Never business-logic-applied. Schema-on-read preserved (or schema promoted but all fields retained). Idempotent ingest by source event ID.

# Ingest Kafka events to Bronze Iceberg
(stream
    .writeStream
    .format("iceberg")
    .outputMode("append")
    .option("path", "s3://lake/bronze/events/")
    .option("checkpointLocation", "s3://lake/checkpoints/bronze_events/")
    .partitionBy("ingest_date")
    .trigger(processingTime="5 minutes")
    .start())

Bronze is the "replayability" layer. If silver/gold ever get corrupted, you can always rebuild from bronze.

Silver — cleaned, conformed, typed, deduped

Business entities emerge. Dimensions and facts in Kimball style. SCD handling lives here.

Gold — consumption-ready, OBT per use case

One gold table per dashboard domain. Materialized from silver. Rebuilt cheaply.

The implicit contract

• Bronze is owned by the ingestion team. Contract: "we will deliver every source event exactly once, preserve schema, and never backfill by overwrite."
• Silver is owned by the platform team. Contract: "conformed dimensions, typed, documented, stable schema."
• Gold is owned by the consumer team. Contract: "you build it, you own it; break it quietly and it's on you."

Medallion + streaming

Works the same way. Bronze = append stream into Iceberg / Delta. Silver = streaming sessionization / dedup / enrichment, written as upsert into Iceberg. Gold = periodic or streaming rollup. Iceberg's row-level DELETE/MERGE (v2) makes this feasible at real latencies.

12. NULL Semantics — The Silent Source of Bugs

NULL in SQL is three-valued logic. NULL = NULL is NULL (not TRUE). x = NULL always yields NULL. This is the source of countless quiet bugs.

NULL in joins

a.x = b.x never matches when either side is NULL. If you want NULL to equal NULL, use:

• a.x IS NOT DISTINCT FROM b.x (Postgres, Spark)
• COALESCE(a.x, '') = COALESCE(b.x, '') (universal but ugly; only works if '' isn't a valid value)
• equal_null(a.x, b.x) (Snowflake)

NULL in aggregations

• COUNT(*) counts all rows.
• COUNT(x) counts non-NULL x.
• SUM, AVG, MIN, MAX ignore NULLs.
• AVG(x) = SUM(x) / COUNT(x) — divides by non-NULL count. Surprising when you expect "nulls treated as zero."
• COUNT(DISTINCT x) counts distinct non-NULL values — so a column with values {1, NULL, NULL} has COUNT DISTINCT 1.

NULL in filters

WHERE x <> 5 excludes rows where x is 5 AND rows where x is NULL. If you want nulls too: WHERE x IS NULL OR x <> 5. Forgetting this is a dominant source of silent data loss.

NULL ordering

• Postgres: NULLs sort LAST by default in ASC, FIRST in DESC. Override with NULLS FIRST / NULLS LAST.
• MySQL: NULLs sort FIRST in ASC.
• Be explicit. Always.

NULL in GROUP BY

NULLs form their own group. GROUP BY country returns one row with country = NULL for all rows missing country. Surface this explicitly in reports or filter before aggregation.

NULLs in window functions

LAG(x) returns NULL before the first row. Use LAG(x, 1, default_value) or COALESCE. LAG(x) IGNORE NULLS (Snowflake/BigQuery) skips NULLs and returns the last non-NULL value — powerful for "last known state" patterns.

-- Last non-null country (for users with intermittent country updates)
SELECT user_id, event_ts,
       LAST_VALUE(country IGNORE NULLS) OVER (
           PARTITION BY user_id ORDER BY event_ts
           ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
       ) AS country_known
FROM events;

Rule: every NULL should be documented

Either NULL means "not yet known" (e.g., churn_date) or "not applicable" (e.g., return_reason on non-returned orders). Document which. Consider sentinel values (unknown codes instead of NULL) when the latter — they force explicit handling downstream.

-- Instead of NULL for "unknown country":
country_code CHAR(2) NOT NULL DEFAULT 'ZZ',   -- ZZ = unknown
-- Reports naturally include "ZZ" as a group rather than silently dropping.

13. Data Contracts & Schema Evolution

A data contract is a guarantee between a producer and consumers about a dataset's schema, semantics, freshness, and quality. Without it, consumers build on sand.

What a contract specifies

# data_contracts/playback_sessions.yaml
id: bronze.playback_sessions
version: 3.1.0
owner: playback-platform@co
schema:
  event_id:        { type: string, required: true, unique: true }
  session_id:      { type: string, required: true }
  user_id:         { type: string, required: true, pii: true }
  title_id:        { type: string, required: true }
  event_ts:        { type: timestamp, required: true, timezone: UTC }
  event_type:      { type: string, required: true, enum: [start, heartbeat, end, error] }
  watch_ms:        { type: long, required: false, constraints: "value >= 0" }
sla:
  freshness: "p99 < 15 minutes after event_ts"
  completeness: ">= 99.5% of upstream client emissions"
  uniqueness: "100% on event_id within 7 days"
tests:
  - name: no_negative_watch_ms
    sql: "SELECT COUNT(*) FROM {{ this }} WHERE watch_ms < 0"
    threshold: 0
breaking_changes_require:
  - major_version_bump
  - 30_day_deprecation_notice
  - consumer_acknowledgment_from: [analytics, ml-platform, product-insights]

Schema evolution rules

Enforcing contracts

Automatic checks in CI:

# contract_check.py
import yaml, pyarrow.parquet as pq

spec = yaml.safe_load(open("data_contracts/playback_sessions.yaml"))
schema = pq.ParquetFile(sample_file).schema_arrow
observed = {f.name: str(f.type) for f in schema}

for field_name, field_spec in spec["schema"].items():
    if field_spec["required"] and field_name not in observed:
        raise ValueError(f"Required field {field_name} missing")
    # Type, enum, constraint checks...

Pair with runtime data quality tests (Great Expectations, Soda, dbt tests) that run on each pipeline execution and block promotion on failure.

Lakehouse-native schema evolution

Iceberg and Delta both support safe evolution:

• Iceberg tracks columns by unique ID, not name. Rename is metadata-only, no rewrite.
• Add column: metadata-only; existing rows treated as NULL.
• Drop column: metadata-only (writes stop emitting it; old files untouched).
• Reorder: metadata-only.
• Type promotion (INT → LONG, FLOAT → DOUBLE): metadata-only, allowed.
• Type narrowing: not allowed.

-- Iceberg
ALTER TABLE silver.playback_sessions ADD COLUMN engagement_score DOUBLE;
ALTER TABLE silver.playback_sessions RENAME COLUMN qoe_score TO quality_score;

14. Modeling Checklist & Anti-Patterns

Before shipping a model:

1. Grain declared and documented? If two engineers could disagree about what a row represents, the model is broken.
2. Are all keys surrogate? Natural keys are attributes, not PKs on dims.
3. Is every measure's additivity documented? Labels in the data catalog; column naming conventions (_point_in_time, _ratio).
4. Is history handled deliberately? SCD type per attribute, not per dim.
5. Are dimensions conformed across facts? Single dim_customer, multiple FKs.
6. Is the date dimension a real table? Not a function call.
7. Are slowly-changing fact attributes modeled as mini-dimensions? Not VARCHAR columns.
8. Are all NULLs meaningful and documented? Sentinel values where applicable.
9. Are timestamps UTC with explicit timezone storage? Local times precomputed for user-local analytics.
10. Does the table have a partition/cluster strategy that prunes typical queries? Default: partition by date, cluster by most-filtered high-cardinality column.
11. Are contracts published and enforced? Schema + SLA + tests in version control, blocked by CI.
12. Can the table be rebuilt from bronze in a reasonable window? If not, you have single-source fragility.

Anti-patterns

• Mixed-grain fact tables. Invisible bug, impossible to audit.
• Natural keys as PKs. Breaks the first time source-system renumbering happens.
• FLOAT for money. Accumulation errors. Use BIGINT cents or DECIMAL(18,2).
• No date dimension. Every report ends up with CASE WHEN EXTRACT(MONTH ...) scattered everywhere.
• TIMESTAMP without timezone. Analytics broken after the next DST change.
• "Latest" table that's actually mutable. A view over SCD2 with WHERE is_current is safe; an actual table that gets UPDATE'd is a mutation-ordering bug farm.
• Soft-delete without a partial index. WHERE deleted = FALSE becomes a full scan on billions of rows.
• EAV (entity-attribute-value) inside a warehouse. (entity_id, attr_name, attr_value) — looks flexible, makes every query a pivot, kills indexability.
• One column per language (name_en, name_fr, name_de, ...). Unbounded schema drift. Normalize to a translations child table.
• Copying source tables verbatim into the warehouse as "the model." Source schema serves source transactions; warehouse schema should serve analytics.
• Logic embedded in views with no materialization. When the dashboard query gets slow, caching becomes a retrofit.
• Dimensions that aren't dimensions. If a column has 8 distinct values, it's a mini-dim candidate, not a VARCHAR fact column.

Next: 02-batch-processing.md — the theory and mechanics of bounded-data processing, idempotency, MERGE patterns, file formats, and partition design.

14. Surrogate Key Strategies in Depth

A surrogate key is an engineered stand-in for a natural key. Choosing the wrong surrogate strategy is a recurring source of production bugs, late-arriving data hell, and expensive re-keying projects two years in.

The three strategies

Sequence (identity column). Monotonic integer issued by the database. Tiny, fast, great for indexing. Fatal weakness: you cannot generate it independently in two systems. If you load facts in Spark and dimensions in Snowflake, you need a round-trip to assign keys. That round-trip is slow, fragile, and forbids truly distributed pipelines.

Hash (deterministic on business key). Apply a collision-resistant hash (SHA-256, xxhash) to the natural key tuple and store the hash. Any system can produce the same key without coordination. Works beautifully for Data Vault and lakehouse pipelines. Trade-off: keys are larger (16–32 bytes) and not human-readable. Birthday-paradox collisions are theoretically possible but practically never occur at data-warehouse scale with 128+ bit hashes.

UUID (random). Nice for unordered distributed generation but terrible for range scans and clustering. Never use random UUIDs as a primary key in a columnar warehouse — they destroy compression and min/max pruning. UUIDv7 (time-sortable) is acceptable if you really want UUID semantics.

Decision table

Hashed surrogate — worked example

-- Snowflake / BigQuery / Spark all agree on this pattern
SELECT
  MD5(CONCAT_WS('|',
    LOWER(TRIM(source_system)),
    CAST(source_id AS STRING),
    CAST(effective_date AS STRING)
  )) AS customer_sk,
  source_system,
  source_id,
  effective_date,
  ...
FROM staging_customer;

Key properties: lowercase + trim on text columns (defensive against inconsistent casing), an explicit delimiter that cannot appear in the payload, and inclusion of the temporal component for SCD Type 2. The same row re-ingested produces the same key — idempotency for free.

15. Bridge Tables and Many-to-Many Dimensions

Not every dimension attaches to a fact with a clean foreign key. A policy can have multiple beneficiaries. An insurance claim can have multiple diagnosis codes. A marketing touch can belong to multiple campaigns. These require bridge tables — and bridge tables are where naive data modelers get caught.

The problem with denormalizing

Tempting shortcut: flatten the many-to-many into a wide fact with diagnosis_code_1, diagnosis_code_2, diagnosis_code_3, …. This works until the day you get a claim with 12 codes. Now you either drop data, create a 20-column wasteland, or reshape the fact — which breaks every downstream consumer. Avoid this pattern.

The bridge table pattern

Three tables: the fact (fct_claim), the dimension (dim_diagnosis), and the bridge (bridge_claim_diagnosis). The bridge stores one row per (claim_sk, diagnosis_sk) with an optional weight column.

CREATE TABLE bridge_claim_diagnosis (
  claim_sk        BIGINT NOT NULL,
  diagnosis_sk    BIGINT NOT NULL,
  is_primary      BOOLEAN NOT NULL,
  weight          DECIMAL(6,4),  -- optional: allocate claim value across codes
  PRIMARY KEY (claim_sk, diagnosis_sk)
);

The weight column is not optional. Without it, you cannot aggregate claim dollars by diagnosis without double-counting. Ten codes on a claim would cause the same dollars to be attributed ten times when joined naively. Interviewers probe for this.

Aggregation safely across a bridge

-- Correct: weighted allocation
SELECT
  d.diagnosis_category,
  SUM(f.claim_amount * b.weight) AS allocated_amount
FROM fct_claim f
JOIN bridge_claim_diagnosis b ON b.claim_sk = f.claim_sk
JOIN dim_diagnosis d           ON d.diagnosis_sk = b.diagnosis_sk
GROUP BY d.diagnosis_category;

-- Wrong: naive join double-counts
SELECT d.diagnosis_category, SUM(f.claim_amount) FROM ...  -- DO NOT

16. Late-Arriving Facts and Dimensions

"Late-arriving" data is the term of art for rows whose business timestamp is in the past but whose arrival timestamp is now. Two distinct cases, each with its own playbook.

Late-arriving facts

A mobile app emits an event on Tuesday but the SDK holds it in local cache due to offline mode, and uploads it Thursday. The fact has a business date of Tuesday. You must decide: does Tuesday's daily aggregate get corrected retroactively, or is the late record attributed to Thursday?

The senior answer is: both tables exist. A "by business date" partition table recomputes Tuesday's totals when the late row arrives. A "by arrival date" partition table leaves Tuesday alone and attributes the row to Thursday. Each serves different consumers — finance cares about business date, operations cares about arrival date — and conflating them is a classic bug.

Late-arriving dimensions

A fact arrives referencing a new customer_id that doesn't exist in dim_customer yet. Three strategies:

1. Inferred member. Insert a placeholder row into dim_customer with the known natural key and NULL attributes. The fact joins successfully. When the full dim row arrives later, update in place (SCD Type 1) or issue a new version (SCD Type 2). Best default.
2. Quarantine. Hold the fact in a quarantine table until the dim row appears. Replay when it does. Use when joining with missing context would be meaningless.
3. Orphan facts. Let the fact land with a NULL dim key. Never recommended — downstream SQL silently drops rows.

17. Accumulating Snapshot Fact Tables

Most facts are transactional (one row per event) or periodic (one row per entity per period). The third, less-known kind is the accumulating snapshot — one row per business process instance, updated in place as the process progresses. The canonical example is an order lifecycle.

Structure

One row per order. Columns for every milestone: placed_ts, paid_ts, shipped_ts, delivered_ts, returned_ts. Plus computed lag columns: days_to_ship, days_to_deliver. The row is created at order placement and UPDATEd at every milestone. NULL columns mean "hasn't happened yet."

CREATE TABLE fct_order_lifecycle (
  order_sk         BIGINT NOT NULL,
  customer_sk      BIGINT NOT NULL,
  placed_ts        TIMESTAMP NOT NULL,
  paid_ts          TIMESTAMP,
  shipped_ts       TIMESTAMP,
  delivered_ts     TIMESTAMP,
  returned_ts      TIMESTAMP,
  days_to_pay      INT,
  days_to_ship     INT,
  days_to_deliver  INT,
  order_amount     DECIMAL(10,2) NOT NULL,
  current_status   VARCHAR(20) NOT NULL
);

Why this fact type exists

Without it, "how long does it take to ship a paid order" requires a self-join on a transactional fact across millions of rows. With accumulating snapshot, it's AVG(days_to_ship). The cost is maintenance complexity: every milestone update has to find and update the right row, ideally via a MERGE on the natural key.

Lakehouse implementation

On Iceberg or Delta, accumulating snapshots are implemented via MERGE INTO. The row is inserted on the first event and updated on every subsequent event. Use a current_status column to track where in the lifecycle the row currently sits — this makes recovery after a failed merge straightforward.

↑ Back to contents

Batch processing is the workhorse of the data warehouse. It's also where the subtle bugs live — the ones that take three weeks to notice and two days to find. This file goes into the theory of bounded-data processing, the mathematics of idempotency, the anatomy of MERGE, the internals of Parquet/ORC, partition design with real math, and how to build pipelines that survive backfills, reruns, and bad data.

1. Mental Model: Bounded vs Unbounded Data (Deep)

A batch job processes bounded data — a finite set of rows with a defined beginning and end. You can scan it, sort it globally, compute the total, and wait for everything before you emit. These are luxuries a streaming job doesn't have.

But "bounded" is a simplification. Most production batch jobs operate on windows of an unbounded stream: "process yesterday's events" is a bounded view of the event firehose. The sealed-ness of that window depends entirely on how late data can arrive.

The three dimensions of boundedness

1. Temporal boundedness — is there a cutoff time after which no more data for the window can arrive?
2. Cardinality boundedness — is the count of rows bounded? (It always is in retrospect, but the job needs to know when it's seen them all.)
3. Completeness boundedness — can you prove all upstream producers have emitted?

Real-world jobs rarely have all three cleanly:

• A daily job on mobile app events: temporally bounded (sealing at midnight + safety window), cardinality uncertain (user activity varies), completeness probabilistic (some clients reconnect days later).
• An S3 daily drop: temporally bounded + cardinality bounded once the drop completes, completeness signal from a _SUCCESS file or a manifest.
• A CDC stream dump for prior day: cardinality bounded by rows in source table; temporal boundedness depends on the CDC tool's lag.

Implications for pipeline design

A truly bounded source (a sealed S3 drop) admits:

• Strict correctness checks: row counts, checksums, column distributions against expected ranges.
• Full refresh fallback: if anything looks off, reprocess from source.

A temporally-bounded-but-late-arriving source admits:

• Safety windows: process event_time ≥ T - 7 days even on the "today" job, so late arrivals are included.
• Periodic reconciliation: a separate weekly job re-processes the last N days to catch late data.
• Grace periods: "seal" day D only after 48h of additional quiet.

Watermark-as-promise, applied to batch

Even batch has watermarks — implicit ones. "Process yesterday" with a 1-hour safety buffer means: I trust that by now() - 1 hour, no events with event_time in [yesterday_start, yesterday_end] will still be arriving. Make this explicit:

def run_daily(dt: date, safety_hours: int = 2):
    """Process dt's events; assume all events for dt have arrived by dt+1day+safety_hours."""
    assert datetime.utcnow() >= datetime.combine(dt + timedelta(days=1), time.min) \
           + timedelta(hours=safety_hours), \
           f"Not safe to process {dt} yet"
    ...

The assert prevents silent correctness loss when backfilling to "today."

Bounded storage, unbounded log

Even a bounded table has an unbounded "log" of changes — the sequence of insertions, updates, deletions over time. Modern lakehouse formats (Iceberg, Delta) expose this log: you can read the table as-of any past snapshot (time travel) or stream the log of changes (Change Data Feed in Delta, incremental reads in Iceberg).

This lets batch jobs consume batch sources as if they were streams:

# Iceberg incremental read — only snapshots since the last run
df = (spark.read
      .format("iceberg")
      .option("start-snapshot-id", last_processed_snapshot)
      .option("end-snapshot-id",   current_snapshot)
      .load("warehouse.bronze.events"))

The line between batch and streaming blurs. Modern advice: unify the code path; only the trigger differs.

2. Anatomy of a Batch Job

A production batch job has these stages, in order. Each can fail; each must be recoverable.

1. Parameter binding — job receives logical_date (the date being processed), not current_date. This is the single most important contract; without it, the job is not idempotent.
2. Source existence check — upstream data exists and is complete. Fail fast if missing (don't produce an empty output that looks successful).
3. Source read — pull from bronze / staging / raw.
4. Deduplication — producers retry, sources double-emit.
5. Conforming — type casting, UTC normalization, null handling, enum validation.
6. Enrichment — joins to dimensions to hydrate descriptors.
7. Transformation — the actual business logic.
8. Validation — row counts, distribution checks, uniqueness assertions. This is the airbag.
9. Write — atomic commit to the target.
10. Publish — signal to downstream consumers (dataset events, XComs, manifest writes).
11. Audit log — row counts in/out, source snapshot IDs, emission latency, job version.

Minimal Python/Spark skeleton

from dataclasses import dataclass
from datetime import date, datetime, timedelta
import logging

log = logging.getLogger(__name__)

@dataclass
class RunContext:
    logical_date: date
    job_version: str
    run_id: str
    spark: "SparkSession"

def run(ctx: RunContext) -> None:
    log.info("job_start", extra={"extra": ctx.__dict__})

    source_exists(ctx)                 # fail fast if upstream missing
    raw = read_source(ctx)             # (1)
    deduped = dedupe(raw)              # (2)
    conformed = conform_types(deduped) # (3)
    enriched = enrich_dimensions(ctx, conformed) # (4)
    output = transform(enriched)       # (5)

    audit = validate(output, ctx)      # (6) — raises on hard-fail
    if audit.soft_warnings:
        notify(audit.soft_warnings)

    write_atomic(output, ctx)          # (7)
    publish_dataset_event(ctx, audit)  # (8)
    log.info("job_done", extra={"extra": {"rows": audit.row_count}})

Each helper should be independently unit-testable; the orchestrator is thin.

The validation gate

This is the most important step and the most often skipped.

@dataclass
class Audit:
    row_count: int
    unique_rate: float
    null_rates: dict[str, float]
    sum_checks: dict[str, int]
    soft_warnings: list[str]

def validate(df, ctx) -> Audit:
    # Hard assertions (raise)
    row_count = df.count()
    if row_count == 0:
        raise AssertionError("Empty output — upstream issue or bug")
    if row_count < ctx.expected_min_rows:
        raise AssertionError(f"Row count {row_count} below min {ctx.expected_min_rows}")
    dup_rate = 1 - df.select("event_id").distinct().count() / row_count
    if dup_rate > 0.001:
        raise AssertionError(f"Uniqueness violated: {dup_rate:.4%} duplicates")

    # Soft warnings (log, notify)
    soft = []
    null_rates = {}
    for col_name in ["user_id", "title_id", "watch_ms"]:
        nr = df.filter(F.col(col_name).isNull()).count() / row_count
        null_rates[col_name] = nr
        if nr > 0.005:  # 0.5% threshold
            soft.append(f"High null rate in {col_name}: {nr:.2%}")

    return Audit(row_count, 1 - dup_rate, null_rates, {}, soft)

Store audit records in a permanent table — historical audit is the tool for diagnosing "what changed overnight."

3. Idempotency — Proofs and Patterns

Definition: running the job twice with the same parameters produces the same output as running it once.

Formally: if f(x) is the job and S_0 is the starting state, then f(f(x))(S_0) = f(x)(S_0). The second run is a no-op relative to the first.

Why it matters

1. Reruns on failure: if your job crashes mid-way, you need to retry without producing duplicates or partial writes.
2. Backfills: reprocessing a historical date must overwrite cleanly, not accumulate.
3. Human error: "I accidentally ran the job twice" shouldn't cause an incident.
4. Distributed retries: orchestrators retry automatically; idempotency prevents them causing damage.

Pattern 1 — Partition overwrite

For append-only outputs partitioned by the job parameter (usually date):

(df.write
  .format("iceberg")
  .mode("overwrite")
  .option("replace-where", f"dt = '{ctx.logical_date}'")
  .saveAsTable("warehouse.silver.playback_sessions"))

The replace-where clause scopes the overwrite to that one partition. Rerun = same partition overwritten with same data = no-op. Iceberg's atomic commit guarantees no reader sees a half-written state.

Safety check: verify df only contains rows for that partition.

assert df.filter(F.col("dt") != ctx.logical_date).count() == 0, \
       "Output contains rows outside target partition — partitioning bug"

Pattern 2 — MERGE by key

For outputs with compound keys and potential updates:

MERGE INTO silver.dim_user AS t
USING stg_user AS s
  ON t.user_id = s.user_id
WHEN MATCHED AND t.updated_at < s.updated_at THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT (...) VALUES (...);

Idempotent because a second run against the same stg_user finds no rows where t.updated_at < s.updated_at (they're equal). Requires a proper version column (updated_at); otherwise you might silently overwrite a newer value with an older one.

Pattern 3 — Hash + upsert

For CDC-like feeds:

MERGE INTO silver.events AS t
USING (
    SELECT *, SHA2(CONCAT_WS('||', event_id, event_ts, payload), 256) AS row_hash
    FROM stg_events
) AS s
  ON t.event_id = s.event_id
WHEN MATCHED AND t.row_hash <> s.row_hash THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT (...) VALUES (...);

Second run: same hashes match, no updates. Idempotent.

Anti-pattern: INSERT ... SELECT without dedup

-- BAD: second run doubles the data
INSERT INTO silver.events SELECT * FROM stg_events;

This is the single most common idempotency violation. Fix:

• Replace with MERGE or partition-overwrite.
• Or: truncate target partition first (but now two statements must be atomic — use Iceberg/Delta semantics).

Anti-pattern: now() in transform

-- BAD: second run produces different output because now() changes
SELECT *, NOW() AS processed_at FROM stg_events;

Fix: use ctx.logical_date or a per-run constant. If you need a "processed_at" column, set it once at the start of the run and pass it in.

Anti-pattern: monotonically increasing keys that differ across runs

# BAD: Spark's monotonically_increasing_id() isn't stable across runs
df.withColumn("key", F.monotonically_increasing_id())

Fix: deterministic hash of natural keys.

df.withColumn("key", F.sha2(F.concat_ws("||", *natural_key_cols), 256))

Testing idempotency

Write a test that runs the job twice and diffs the output:

def test_idempotent_run(tmp_table, test_data):
    run(RunContext(logical_date=date(2026,4,10), ...))
    snapshot1 = spark.table(tmp_table).collect()
    run(RunContext(logical_date=date(2026,4,10), ...))
    snapshot2 = spark.table(tmp_table).collect()
    assert snapshot1 == snapshot2, "Output differs between runs — not idempotent"

This catches now() leaks, non-deterministic ordering, and accidental row duplication.

4. MERGE Under the Hood

MERGE (aka UPSERT) looks simple; it's anything but. Understanding its execution is essential to reasoning about performance and correctness.

The logical model

MERGE INTO target t
USING source s
  ON t.key = s.key
WHEN MATCHED AND <cond> THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...
WHEN MATCHED AND <cond2> THEN DELETE
WHEN NOT MATCHED BY SOURCE THEN DELETE;   -- supported by some engines

Conceptually: for every (t, s) pair where t.key = s.key, apply the first matching WHEN MATCHED clause. For every s with no match in t, apply WHEN NOT MATCHED. For every t with no match in s (when WHEN NOT MATCHED BY SOURCE), apply that clause.

The multi-match rule

If a single target row matches multiple source rows, behavior is undefined and most engines error:

ERROR: Cannot perform MERGE as multiple source rows matched and attempted to modify
       the same target row.

Fix: deduplicate the source to a single row per join key before the MERGE.

MERGE INTO target t
USING (
    SELECT DISTINCT ON (user_id) *           -- Postgres
    FROM stg_user
    ORDER BY user_id, updated_at DESC
) s
ON t.user_id = s.user_id
...

Or using window functions (portable):

USING (
    SELECT * FROM (
        SELECT *, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY updated_at DESC) rn
        FROM stg_user
    ) WHERE rn = 1
) s

Execution on lakehouse (Iceberg/Delta)

MERGE on a lakehouse is not an in-place update. Files are immutable. The steps:

1. Scan target to find files touched by matched rows.
2. Scan source and join against target on key.
3. For each target file touched, read it fully, apply the merge, write a new file with updated/inserted rows.
4. Commit: swap the manifest to reference new files instead of old.
5. Old files remain (for time travel) until vacuum.

Implications:

• Write amplification: updating 1% of rows in a 10 GB file still rewrites the whole 10 GB file. This is why small, well-sized files matter — bigger files = more write amplification per update.
• Merge-on-read vs copy-on-write: Iceberg v2 and Delta support "merge-on-read" mode where deletes/updates are stored as separate delete files (positional or equality deletes), merged at read time. Cheaper writes, more expensive reads until compaction.
• Compaction is essential. Without periodic OPTIMIZE / rewrite_data_files, merge-on-read tables slow down over time.

Copy-on-write vs merge-on-read

Copy-on-write (default in Iceberg v1, Delta default):
  Update → rewrite touched files entirely.
  Pros: fast reads, no merge at query time.
  Cons: slow writes, high write amplification.
  Use: update-light workloads (SCD2, daily overwrites).

Merge-on-read (Iceberg v2, Delta with deletion vectors):
  Update → write delete file + new rows file.
  Pros: fast writes.
  Cons: reads must apply delete vectors; slower until compacted.
  Use: update-heavy workloads (CDC, streaming upserts).

Configure per table based on expected update pattern.

Partition-aware MERGE

If the target is partitioned and the merge key lets the planner prune partitions, only those are touched. Make sure the partition column appears in the ON clause:

-- Good: planner prunes partitions where dt doesn't appear in source
MERGE INTO target t
USING source s
ON t.user_id = s.user_id AND t.dt = s.dt   -- includes partition col
...

Without the partition predicate in the ON clause, the engine scans all partitions.

Spark / Delta MERGE performance

from delta.tables import DeltaTable

# Enable deletion vectors (merge-on-read) for this table
spark.sql("""
    ALTER TABLE silver.events SET TBLPROPERTIES (
        'delta.enableDeletionVectors' = 'true'
    )
""")

t = DeltaTable.forName(spark, "silver.events")
(t.alias("t")
    .merge(
        source=stg.alias("s"),
        condition="t.event_id = s.event_id AND t.dt = s.dt"
    )
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

Tips:

• Narrow the MERGE with a partition predicate if possible.
• Use spark.databricks.delta.merge.repartitionBeforeWrite.enabled = true to redistribute writes for better parallelism.
• Monitor numOutputRows vs numTargetRowsUpdated — if you're rewriting 100% for a 1% update, reconsider file sizing.

5. Incremental Processing Patterns

Full refresh is simple but expensive above ~100 GB. Incremental processing is cheap but introduces correctness traps.

Pattern 1 — High watermark incremental

Track the max processed timestamp per source; next run pulls everything since.

def get_high_watermark() -> datetime:
    return spark.sql("""
        SELECT MAX(event_ts) FROM silver.events
    """).collect()[0][0] or datetime.min

def run_incremental():
    hwm = get_high_watermark()
    new_rows = spark.read.table("bronze.events") \
                    .filter(F.col("event_ts") > hwm)
    # process...
    new_rows.write.mode("append").saveAsTable("silver.events")

Correctness traps:

• Late-arriving data with event_ts < hwm never gets processed. Fix: widen the window (process event_ts > hwm - safety_window, then dedupe).
• Clock skew across producers can push HWM forward too fast. Fix: use event_ts not processing_ts.
• A failed run that half-updated the target leaves the HWM in a wrong state. Fix: watermark is derived from the committed target, not stored separately.

Pattern 2 — Partition-based incremental

The target has daily partitions; each run processes one partition's worth of source data.

def run_incremental(dt: date):
    source_for_day = (spark.read.table("bronze.events")
        .filter(F.col("dt") == dt))   # safety: always one partition
    processed = transform(source_for_day)
    (processed.write
        .format("iceberg")
        .mode("overwrite")
        .option("replace-where", f"dt = '{dt}'")
        .saveAsTable("silver.events"))

Cleaner than HWM — each partition is an atomic unit. Reruns are safe. Backfills loop over dates.

Pattern 3 — Snapshot-diff incremental (Iceberg / Delta)

Read only the changes to a source table since the last processed snapshot.

# Iceberg
df_changes = (spark.read
    .format("iceberg")
    .option("start-snapshot-id", last_snapshot)
    .option("end-snapshot-id",   current_snapshot)
    .load("warehouse.bronze.events"))

# Delta: Change Data Feed (CDF)
df_changes = (spark.read
    .format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", last_version)
    .option("endingVersion",   current_version)
    .load("/path/to/bronze/events"))
# Rows include _change_type = 'insert' | 'update_preimage' | 'update_postimage' | 'delete'

Much cheaper than scanning the whole table when most data is unchanged. Used for streaming-from-batch and silver → gold refresh.

Pattern 4 — Merge-on-arrival (streaming batch)

For sources that arrive as a stream but are processed in micro-batches (Spark Structured Streaming with Trigger.AvailableNow):

(spark.readStream
    .format("iceberg")
    .load("warehouse.bronze.events")
    .writeStream
    .format("iceberg")
    .option("path", "warehouse.silver.events")
    .option("checkpointLocation", "s3://.../ckpt/silver_events/")
    .trigger(availableNow=True)    # process whatever's available, then stop
    .start()
    .awaitTermination())

Best of both: streaming semantics (state, checkpoints, exactly-once), batch cadence (scheduled run, then stop).

6. Backfills — Design, Safety, and Throttling

A backfill is how pipelines prove themselves. The healthy pattern:

1. Parameterized job (takes logical_date).
2. Idempotent writes (partition overwrite or MERGE).
3. Separate orchestrator (dedicated DAG or job) that doesn't block the daily schedule.
4. Rate limiting to avoid flooding upstream or spiking cost.
5. Monitoring: backfill progress, failure rate, error logs.

Backfill driver script

def backfill(start: date, end: date, parallelism: int = 4, throttle_s: int = 30):
    dates = [start + timedelta(days=i) for i in range((end - start).days + 1)]
    with ThreadPoolExecutor(max_workers=parallelism) as pool:
        futures = {}
        for i, d in enumerate(dates):
            futures[pool.submit(run_daily, d)] = d
            time.sleep(throttle_s / parallelism)   # stagger starts
        for f in as_completed(futures):
            d = futures[f]
            try:
                f.result()
                log.info(f"backfilled {d}")
            except Exception as e:
                log.error(f"backfill failed for {d}: {e}")
                # Decide: halt entire backfill, or continue and report?

Key knobs:

• Parallelism: run N days simultaneously. Too high = overwhelms source / cluster.
• Throttle: pause between submissions. Prevents thundering herd.
• Error policy: fail fast vs continue. Depends on whether days are independent.

Airflow dynamic task mapping for backfills

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    schedule=None,            # triggered manually
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_tasks=8,        # concurrency limit
)
def backfill_playback():
    @task
    def list_dates(start: str, end: str) -> list[str]:
        s = datetime.fromisoformat(start).date()
        e = datetime.fromisoformat(end).date()
        return [(s + timedelta(days=i)).isoformat() for i in range((e-s).days+1)]

    @task(retries=2)
    def run_one(dt: str):
        run_daily_for(dt)

    run_one.expand(dt=list_dates("{{ params.start }}", "{{ params.end }}"))

backfill_playback()

Trigger with airflow dags trigger backfill_playback -p '{"start":"2026-01-01","end":"2026-02-01"}'.

Backfill correctness

• If the current pipeline version differs from the one that produced the historical data, backfill produces different results. Document this; version-tag the output.
• Upstream sources may have had different schemas historically; backfill code must handle all historical shapes OR start from a normalized bronze.
• Dimensions are SCD2: a fact backfilled today joins to dim versions that were current on the backfill date, not today.

Cost-aware backfills

A naive "backfill one year of events" can cost $50k on a warehouse. Tactics:

• Sampling: backfill 10% first; verify output shape; extrapolate cost; decide.
• Coarse-grained backfill: backfill weekly rollups first; fine-grained only where needed.
• Use a dedicated cluster/warehouse: don't let the backfill contend with production.
• Cold-path data: older data goes to cheaper storage tiers (S3 Glacier, Snowflake external tables).

7. File Format Internals — Parquet, ORC, Avro

Parquet (columnar, analytic)

Layout:

File
├── Magic bytes "PAR1"
├── Row Group 1
│   ├── Column Chunk: col_a
│   │   ├── Page 1 (data + RLE/dict encoding)
│   │   ├── Page 2
│   │   └── ...
│   ├── Column Chunk: col_b
│   └── ...
├── Row Group 2
├── ...
├── File Metadata (schema, row group locations, column stats)
└── Magic bytes "PAR1"

• Row group size: 128 MB default. Smaller = finer pruning but more overhead per file. Larger = more efficient scans but coarser filtering.
• Page size: 1 MB default. The unit of decompression.
• Dictionary page: encode low-cardinality columns as ints pointing to a dictionary. Spark writes dictionary pages up to 1 MB; then falls back to plain encoding.

Statistics available:

• Per row group, per column: min, max, null_count, distinct_count (optional).
• Bloom filters (Parquet 2.5+, optional): probabilistic set membership per column chunk.

Encodings:

• PLAIN — raw values, no compression.
• RLE_DICTIONARY — RLE of dictionary IDs. Default for most columns.
• DELTA_BINARY_PACKED — delta encoding. Good for sorted integers / timestamps.
• DELTA_BYTE_ARRAY — delta encoding for strings with common prefixes.

Compression:

• SNAPPY — fast, moderate compression. Default.
• ZSTD — slower, better compression (10-20% smaller than SNAPPY). Preferred in 2026.
• GZIP — very slow, best compression. Rarely used.
• LZ4 — fastest decompression.

Reading pattern (what the engine does):

1. Open file, seek to footer, read metadata.
2. Apply filters against per-row-group stats; skip non-matching row groups.
3. For each surviving row group, for each needed column, seek to column chunk.
4. Decode pages; apply predicates again at page level (if dictionary-filter eligible).
5. Assemble rows.

This is why selecting only needed columns and filtering on stat-having columns is so fast in Parquet.

ORC (columnar, Hive-era)

Similar to Parquet but:

• Stripe = ORC's row group (64 MB default).
• Richer lightweight indexes (per-10k-row bloom filters, not just per-stripe).
• Better complex-type handling historically (structs, maps).
• Compresses slightly better (ZSTD/ZLIB) due to more aggressive type-aware encoding.

Most modern lakehouses use Parquet for ecosystem reasons. ORC dominant in traditional Hadoop.

Avro (row-oriented, streaming)

Row-oriented format. Bad for analytics (must read full rows); good for:

• Streaming source data: Kafka topics often serialize records as Avro with a schema registry.
• Backup / archival: faster to produce, less CPU to write.
• Schema evolution: Avro has a rich schema resolution model (reader schema vs writer schema, nullable union types).

# Writing Avro
df.write.format("avro").save("s3://backup/events/")

# Reading with explicit schema
import avro.schema
schema = avro.schema.parse(open("events.avsc").read())

Format choice

• Analytic tables: Parquet (lakehouse default) or ORC (Hive legacy).
• Streaming bronze (Kafka → lake): Avro in-flight (schema registry), Parquet on landing.
• Archival / backup: Avro or gzipped JSON.
• Row-level state: neither — use a row store (Postgres, RocksDB).

8. Partition Design Math

Partition choice is the #1 physical layout decision in a lakehouse. Bad partitioning wastes more compute than any other mistake.

The cardinality sweet spot

Rule: each partition should hold hundreds of MB to a few GB of uncompressed data, with at least a few hundred partitions total for a large table and not more than ~tens of thousands.

Math:

• Row group size target: 128 MB.
• Files per partition: 1–10 for good parallelism without small-file overhead.
• → Partition size target: 128 MB – 1.3 GB.

For a table with 10 TB of data:

• 10 TB / 500 MB per partition ≈ 20,000 partitions. Upper edge.
• 10 TB / 5 GB per partition ≈ 2,000 partitions. Nicer.

If the natural partition column (e.g., dt) gives you 10,000 days — that's 10,000 partitions, ~1 GB each.

If the natural partition column gives you 10M users — that's 10M partitions of tiny data each. Disaster. Repartition by a coarser key (day + country, or user_id % 1000).

Partition evolution

Iceberg supports partition evolution: change the partition spec without rewriting data. Old data keeps its partitioning; new writes use the new spec.

-- Started with daily partitions
ALTER TABLE silver.events SET PARTITION SPEC (days(event_ts), bucket(16, user_id));
-- Old daily partitions survive; new writes use day+user_bucket

This is huge. In traditional Hive, changing partitioning meant rewriting the entire table.

Partition pruning — what actually happens

Query: SELECT * FROM silver.events WHERE dt = '2026-04-19' AND country = 'US'.

1. Planner reads table metadata → sees partition spec days(event_ts) → partition column dt.
2. Planner matches filter dt = '2026-04-19' → determines relevant partition(s).
3. Planner reads manifest files for matching partitions only.
4. Manifests list data files with per-column stats (min/max of country).
5. Planner prunes files where country stats exclude 'US'.
6. Spark launches tasks only against surviving files.

What breaks pruning:

• Function on partition column: WHERE DATE(event_ts) = '2026-04-19' — planner can't prove this matches partition days(event_ts)=2026-04-19. Rewrite: WHERE event_ts >= '2026-04-19' AND event_ts < '2026-04-20'.
• Iceberg's hidden partitioning fixes this — days(event_ts) is an implicit partition derivation, and filters on event_ts directly are pruned.
• Implicit type casts: WHERE dt = '20260419' (string) vs partition column is DATE — some engines won't prune. Match types.
• UDFs in predicates: opaque; never pruned.

Hidden partitioning (Iceberg's killer feature)

Hive-style: the partition column is user-managed. You must always write WHERE dt = '...' matching the partition value exactly.

Iceberg: partition derived from source column. User writes WHERE event_ts BETWEEN ...; Iceberg transparently computes days(event_ts) and prunes.

-- Iceberg
CREATE TABLE silver.events (
    event_id STRING,
    event_ts TIMESTAMP,
    user_id STRING,
    ...
)
USING iceberg
PARTITIONED BY (days(event_ts), bucket(16, user_id));

-- Query naturally uses event_ts (no dt column visible)
SELECT COUNT(*) FROM silver.events
WHERE event_ts BETWEEN '2026-04-19' AND '2026-04-20'
  AND user_id = 'abc';
-- Prunes: days(event_ts) matches 2026-04-19 partition
-- Prunes: bucket(16, 'abc') matches one of 16 buckets

Clustering (sort within partition)

Even with good partitioning, within a partition the data order matters. Clustering (Snowflake), z-ordering (Delta), or sort-order (Iceberg) physically sorts rows so filters prune at the file / row-group level.

-- Iceberg: write new data sorted by user_id within each partition
ALTER TABLE silver.events WRITE ORDERED BY user_id;

-- Delta: z-order after writes (multi-column space-filling curve)
OPTIMIZE silver.events ZORDER BY (user_id, country);

-- Snowflake: clustering key (automatic maintenance)
ALTER TABLE silver.events CLUSTER BY (user_id);

When to cluster:

• You frequently filter on a specific high-cardinality column (e.g., user_id).
• That column is too high-cardinality to partition on directly.
• Query patterns are read-heavy; the sort amortizes.

Z-order packs rows with similar values across multiple columns. Best when queries filter on a pair of columns unpredictably — e.g., sometimes user_id, sometimes country, sometimes both. Single-column sort is better when there's one dominant filter.

Bucket partitioning

Divide rows into N buckets by hash(key) % N. Useful when:

• Key is high-cardinality (can't partition directly).
• You want predictable file sizes (each bucket has roughly 1/N of data).
• Joins on that key can use bucketed joins (skip shuffle if both sides are bucketed identically).

CREATE TABLE silver.events ( ... )
USING iceberg
PARTITIONED BY (days(event_ts), bucket(32, user_id));

32 buckets × 365 daily partitions = 11,680 partitions for a year. Each partition covers 1_day × (1/32 of users). A query filtering on user_id = 'abc' prunes to 1 day × 1 bucket = small.

9. Small Files and Compaction

The small files problem: tiny files kill read performance because each file incurs fixed overhead (open, footer read, metadata parse). A table with millions of tiny files is slower than one with thousands of right-sized files, despite holding the same data.

Causes of small files

• High write parallelism + low data volume. 200 Spark partitions × 200 hourly runs = 40,000 files even if each run is small.
• Streaming micro-batches with short triggers. 1-minute triggers for a low-volume stream: ~525,000 files per year per topic.
• Skewed partitioning. A partition with lots of data gets sub-partitioned; one with little gets one tiny file.

Prevention at write time

• df.coalesce(N) before write: reduces file count without shuffle. Use N = target_data_size / target_file_size.
• df.repartition(N, ...) before write: shuffles to exactly N partitions. Costs a shuffle but guarantees even file sizes.
• Spark AQE: spark.sql.adaptive.advisoryPartitionSizeInBytes = 128MB — AQE coalesces output partitions to hit this target.
• Iceberg write.target-file-size-bytes table property.

df.write \
  .format("iceberg") \
  .option("target-file-size-bytes", 134217728) \  # 128 MB
  .mode("append") \
  .saveAsTable("silver.events")

Compaction (post-write)

Running OPTIMIZE or rewrite_data_files periodically rewrites many small files into fewer large ones.

-- Iceberg: compact files smaller than 100 MB, combining to ~512 MB
CALL system.rewrite_data_files(
    table => 'silver.events',
    options => map(
        'target-file-size-bytes', '536870912',
        'min-file-size-bytes',    '104857600'
    )
);

-- Delta
OPTIMIZE silver.events;

Schedule: daily for streaming-ingested tables, weekly for batch-ingested.

Garbage collection (expiring snapshots)

Iceberg/Delta keep old snapshots for time travel. Expire them to reclaim storage:

-- Iceberg: expire snapshots older than 7 days, keep at least 5 most recent
CALL system.expire_snapshots(
    table => 'silver.events',
    older_than => TIMESTAMP '2026-04-12 00:00:00',
    retain_last => 5
);

-- Delta
VACUUM silver.events RETAIN 168 HOURS;   -- 7 days

Caution: vacuuming kills time-travel for data older than the retention window. If compliance requires 7-year retention, configure accordingly.

10. CDC (Change Data Capture) Patterns

Source OLTP database → analytics warehouse. The canonical patterns:

Full snapshot (baseline)

Daily full dump of the source table. Simple, correct, expensive. Viable up to ~tens of GB.

# JDBC to Iceberg (full reload)
(spark.read
    .format("jdbc")
    .option("url", "jdbc:postgresql://...")
    .option("dbtable", "public.orders")
    .option("partitionColumn", "order_id")
    .option("numPartitions", 20)
    .option("lowerBound", "1")
    .option("upperBound", "100000000")
    .load()
    .write
    .format("iceberg")
    .mode("overwrite")
    .saveAsTable("bronze.orders"))

Pros: guaranteed correctness, no state. Cons: expensive at scale, lag = 1 day.

Incremental snapshot (changed-rows-only)

Source table has updated_at; pull rows where updated_at > last_pull.

hwm = get_high_watermark("bronze.orders", "updated_at")
(spark.read
    .format("jdbc")
    .option("dbtable", f"(SELECT * FROM orders WHERE updated_at > '{hwm}') AS t")
    ...
    .write
    .format("delta")
    .mode("append")
    .saveAsTable("bronze.orders_changes"))

# Upsert into bronze.orders using MERGE

Pros: cheaper than full reload. Cons: misses hard deletes (row gone from source with no marker); relies on accurate updated_at.

Log-based CDC (Debezium, Fivetran, AWS DMS)

Read from the database's write-ahead log (Postgres logical replication, MySQL binlog, SQL Server CDC). Emit Kafka messages per row change, with operation type:

{
  "before": {"order_id": 42, "status": "paid",   "total": 100},
  "after":  {"order_id": 42, "status": "shipped", "total": 100},
  "op": "u",
  "ts_ms": 1713567890000,
  "source": {"schema": "public", "table": "orders", "lsn": "0/30A3B8"}
}

Ops: c (create/insert), u (update), d (delete), r (snapshot read).

Pros:

• Sub-second lag.
• Hard deletes captured.
• Exact change history, not just "current state."
• Works without touching the source.

Cons:

• Operationally complex: Kafka, Debezium, schema registry.
• Requires database-side logging (WAL) configured correctly.
• Schema drift at source breaks pipelines.

Upserting CDC into a lakehouse

cdc_stream = (spark.readStream
    .format("kafka")
    .option("subscribe", "cdc.public.orders")
    ...
    .load())

# Parse Debezium envelope
parsed = cdc_stream.select(
    F.from_json(F.col("value").cast("string"), debezium_schema).alias("env")
).select(
    "env.op", "env.after", "env.before", "env.ts_ms"
)

# Apply to target with forEachBatch + MERGE
def merge_batch(batch_df, batch_id):
    from delta.tables import DeltaTable
    t = DeltaTable.forName(spark, "silver.orders")
    (t.alias("t")
        .merge(
            batch_df.alias("s"),
            "t.order_id = s.after.order_id OR (s.op = 'd' AND t.order_id = s.before.order_id)"
        )
        .whenMatchedDelete(condition="s.op = 'd'")
        .whenMatchedUpdateAll(condition="s.op = 'u' AND s.ts_ms > t.cdc_ts_ms")
        .whenNotMatchedInsert(
            condition="s.op IN ('c','r')",
            values={
                "order_id": "s.after.order_id",
                "status":   "s.after.status",
                ...
                "cdc_ts_ms": "s.ts_ms"
            }
        )
        .execute())

(parsed.writeStream
    .foreachBatch(merge_batch)
    .option("checkpointLocation", "s3://.../ckpt/orders_cdc/")
    .trigger(processingTime="30 seconds")
    .start())

The cdc_ts_ms > t.cdc_ts_ms guard prevents out-of-order CDC events from overwriting a newer state with an older one.

CDC + SCD2

For downstream dim tables: the stream of CDC events drives SCD2 insertion of new rows. Every update event closes the current row and inserts a new one; every delete marks the current row as invalidated.

11. Data Quality in Batch Pipelines

Quality checks live alongside logic — not "if we have time later."

Taxonomy of checks

dbt tests (declarative)

# models/silver/playback_sessions.yml
version: 2
models:
  - name: playback_sessions
    description: "Sessionized playback events"
    columns:
      - name: session_id
        tests:
          - unique
          - not_null
      - name: user_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_user')
              field: user_id
      - name: watch_ms
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 86400000   # 24h in ms
    tests:
      - dbt_utils.recency:
          datepart: hour
          field: max_end_ts
          interval: 2

Great Expectations (imperative + metadata)

import great_expectations as gx

context = gx.get_context()
ds = context.sources.add_pandas("pd")
asset = ds.add_parquet_asset(
    "playback_sessions",
    filepath_or_buffer="s3://lake/silver/playback_sessions/dt=2026-04-19/*.parquet"
)

batch = asset.build_batch_request()
validator = context.get_validator(batch_request=batch)

validator.expect_column_values_to_not_be_null("session_id")
validator.expect_column_values_to_be_unique("session_id")
validator.expect_column_values_to_be_between("watch_ms", min_value=0, max_value=86_400_000)
validator.expect_column_mean_to_be_between("watch_ms", min_value=60_000, max_value=3_600_000)

validator.save_expectation_suite("playback_sessions.suite")
result = validator.validate()
if not result.success:
    raise ValueError(f"DQ failed: {result}")

Soda Core (YAML-first)

checks for silver.playback_sessions:
  - row_count > 1000000
  - missing_count(session_id) = 0
  - duplicate_count(session_id) = 0
  - values in (device_type) must be in ['tv','mobile','tablet','web','other']
  - freshness(end_ts) < 1h
  - anomaly score for row_count < 3

Designing "what to check"

• Every PK column: not null, unique.
• Every FK: relationships to parent (or tolerate N% unresolved with a warning).
• Every measure: physically plausible range (no negatives on watch time, etc.).
• Every timestamp: within expected date range (catch 1970-01-01 defaults).
• Every enum column: values in allowed set.
• Daily volumes: within ±20% of 7-day rolling average; alert on anomalies.

Over-checking is cheap; under-checking is expensive.

12. Orchestration Patterns for Batch

The simple scheduled DAG

One task per stage, linear dependencies, cron schedule.

extract → dedupe → transform → validate → write → publish

Best for: small, well-understood pipelines.

Asset-based orchestration (Dagster)

Instead of tasks-with-dependencies, declare the datasets (assets) and their producers. Dagster figures out the dependency graph.

from dagster import asset

@asset
def bronze_events(context):
    return pull_from_kafka(context)

@asset
def silver_events(bronze_events):
    return transform(bronze_events)

@asset
def gold_daily_plays(silver_events):
    return aggregate(silver_events)

Materializing gold_daily_plays rebuilds upstream as needed.

Sensor-based triggering

Instead of cron, trigger when upstream data arrives.

# Airflow
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait = S3KeySensor(
    task_id="wait_for_upstream",
    bucket_key="s3://upstream/events/{{ ds }}/_SUCCESS",
    timeout=60 * 60 * 4,
    poke_interval=300,
)

For lakehouse tables: dataset-based scheduling (Airflow 2.4+) reacts when the upstream table is updated.

Backfill separation

Never use the same DAG for daily and backfill. Daily runs should not contend with backfills. Backfill DAG:

• Dedicated compute pool.
• Lower priority.
• Separate alerting (backfill failures rarely need 3am pages).

SLA management

Every table should have:

• Freshness SLO — "updated within 1h of upstream seal."
• Completeness SLO — "≥99% of expected row count."
• Correctness SLO — "DQ tests green on every run."

Expose as metrics; alert on violations with an on-call rotation.

# Airflow SLA miss callback
def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    notify_on_call(...)

default_args = {
    "sla": timedelta(hours=1),
    "sla_miss_callback": sla_miss_callback,
}

Next: 03-streaming-processing.md — unbounded data, event-time processing, watermark math, Flink and Kafka internals, exactly-once proofs.

13. Dependency Graphs and Critical Path Analysis

A batch "pipeline" at scale is never a chain — it's a directed acyclic graph of hundreds of tasks with cross-dataset dependencies. Two questions dominate: when will the whole graph finish, and which node, if it slips, slips the whole graph? These are critical-path questions.

Modeling the graph

Each node is a task with a duration. Each edge expresses "task B cannot start until task A succeeds." Total graph completion time is the longest-path sum from any source to any sink. That longest path is the critical path.

Nodes off the critical path have slack — the amount they can slip without affecting total completion time. Nodes on the critical path have zero slack. A single extra minute on any critical-path task adds a minute to the whole pipeline.

Why this matters for interviews

"Our daily pipeline finishes at 6am but the business wants it by 4am — how do you speed it up?" Wrong answer: "I'd optimize the slowest job." Right answer: "I'd find the critical path and optimize the longest task on it. Optimizing off-path tasks won't move completion time at all."

The Airflow / Dagster reality

Most orchestrators expose task duration logs. Extract them, build the adjacency matrix, compute longest-path via topological sort + dynamic programming. O(V+E). For a 500-task graph, runs in sub-second and tells you exactly which five tasks you must optimize.

# Simplified: compute critical path duration
def critical_path(nodes, edges, duration):
    # topological sort
    indeg = {n: 0 for n in nodes}
    for a, b in edges: indeg[b] += 1
    order, q = [], [n for n,d in indeg.items() if d == 0]
    while q:
        n = q.pop(0); order.append(n)
        for a, b in edges:
            if a == n:
                indeg[b] -= 1
                if indeg[b] == 0: q.append(b)
    # DP: earliest finish time
    finish = {n: duration[n] for n in nodes}
    for n in order:
        for a, b in edges:
            if a == n:
                finish[b] = max(finish[b], finish[n] + duration[b])
    return max(finish.values())

14. The Batch Cost Model

Senior candidates should be able to estimate batch cost without invoking the cloud-provider pricing page. The model has four terms.

The four costs

1. Storage. GB-months × $/GB-month. Flat. For S3 Standard, ~$0.023/GB-month. For Glacier, ~1/10 of that.
2. Compute. Core-hours × $/core-hour. EC2 spot typically $0.01–0.02 per vCPU-hour; on-demand 3x that.
3. Shuffle / intermediate I/O. Bytes written to local disk or the shuffle service. On ephemeral disk, free but capacity-limited. On remote shuffle (cloud managed), charged per GB.
4. Egress. Cross-region or internet egress. The silent killer. $0.02–0.09 per GB depending on direction. Can dwarf compute cost for lift-and-shift workloads.

Worked estimate — 10 TB daily batch

Input 10 TB. One Spark job with a 3-way join producing 2 TB output. Shuffle amplification typically 2–4x input — say 30 TB of shuffle. Runtime 40 min on 100 vCPUs.

• Compute: 100 vCPU × 0.67 hour × $0.02 = $1.33
• Storage (same-region S3, input rereadable): negligible incremental
• Shuffle: 30 TB local disk, free
• Egress: zero if same region, ~$900 if cross-region output

The answer "this job costs roughly $1.50 same-region, $900+ cross-region" is the senior-level answer. Notice the 600x swing just on egress — that's why "lift-and-shift to another region" is so often the wrong design move.

15. Parallelism Math and the Skew Tax

Perfectly parallel work distributes across N workers with speedup ≈ N. Real work rarely achieves this. Two sources of loss: sequential fraction (Amdahl's law) and skew (one partition is fatter than the rest).

The skew tax, quantified

If you partition 1 TB of work into 100 evenly-sized pieces of 10 GB each, total wall-clock time is ≈ (10 GB work / per-worker throughput). If instead one partition is 500 GB and the other 99 average 5 GB, total wall-clock is bounded below by 500 GB / throughput — 50x slower than the balanced case, despite the same total work. That's the skew tax.

Detecting skew ahead of time

• Compute partition size distribution: SELECT part_key, COUNT(*), SUM(bytes) FROM … GROUP BY 1 ORDER BY 2 DESC LIMIT 20. If the top partition is 10x the median, you have skew.
• Coefficient of variation (stddev / mean) on partition sizes. Above 0.5 = significant skew.

Mitigations

• Salting. Suffix the skewed key with a small random integer (key + '_' + RAND(10)), join, then aggregate up. Fixes the shuffle but doubles join complexity.
• Broadcast. If the other side is small, broadcast it and avoid shuffling the skewed key entirely.
• Isolate the hotspot. Split the skewed keys into a separate job with different parallelism.
• AQE. Spark's adaptive skew-join detects skewed partitions at runtime and splits them. Configure spark.sql.adaptive.skewJoin.enabled=true, know what skewedPartitionThresholdInBytes controls.

16. Checkpoint and Restart Semantics

Batch jobs are not point-in-time atomic. A 6-hour job that fails at hour 4 either resumes from a checkpoint (if designed for it) or restarts from zero. The senior design decision is which.

Three restart strategies

1. Fully idempotent restart from start. Every job is safe to rerun end-to-end. Requires deterministic inputs and idempotent sinks (MERGE on business key). Simple; operationally resilient. Cost: wasted compute on every failure.
2. Task-level checkpointing. Orchestrator records task completion; restart resumes at the failed task. Works for DAG-structured jobs. Requires each task's output to be written atomically (typically via temp-then-rename).
3. Intra-task checkpointing. A long task persists progress state (e.g., partition cursor) and resumes from it. Necessary for multi-hour jobs. Expensive to implement right; usually worth it only for jobs that exceed ~2 hours.

The atomic-write discipline

Every job output must be all-or-nothing. The pattern: write to a temp location, validate, then atomic rename or commit. Never "write in place and pray." Lakehouse formats (Iceberg, Delta) get you atomic commits for free. Hive partitioned tables do not — use _SUCCESS markers and write to a staging path first.

The "restart at noon" drill

Interviewers love to ask: "Yesterday's job failed at 11pm. You rerun at noon today. What happens downstream?" The senior answer enumerates: which tables get rewritten, which downstream caches invalidate, which dashboards see a stale→fresh transition, which SLAs trip. If you can't answer this for your own pipelines, you don't own them operationally — no matter what the org chart says.

17. Partition Key Decisions by Domain

Choosing a partition key is the single highest-leverage physical-design decision in a lakehouse or warehouse table. A good choice makes common queries prune 99% of the data; a bad one produces billion-row scans for every dashboard. Below is a pattern-match table of typical domains, the natural partition key, the access patterns that justify it, and the most common wrong choice candidates reach for first.

Sizing heuristic

Aim for partitions in the ~100 MB – ~10 GB range after compaction. Below that, metadata overhead dominates. Above it, a single consumer reading one partition loses parallelism. For a daily-partitioned 1 TB/day fact, that's ideal. For a 100 GB/day fact, daily is still fine. For a 10 TB/day fact, move to hourly.

The bucketed-id trick

When you have a high-cardinality key (user_id, device_id) that users filter on but can't partition on directly, hash it into a fixed number of buckets and use that as a secondary partition. The query can push the bucket predicate down (WHERE user_bucket = hash(user_id) % 32), pruning ~97% of partitions while keeping each partition reasonable size. This is how Iceberg's hidden partitioning works under the hood for bucket transforms.

-- Iceberg: hidden bucketing — no user code changes
CREATE TABLE events (
  event_date DATE,
  user_id BIGINT,
  event_name STRING,
  ...
)
USING ICEBERG
PARTITIONED BY (event_date, bucket(32, user_id));

-- Queries that filter on user_id automatically prune 31/32 buckets
SELECT * FROM events
WHERE event_date = '2026-04-20' AND user_id = 12345;

18. NULL Semantics — Examples by Domain

NULL is never a single thing. It carries different meanings across columns and domains, and conflating them is a root cause of subtle downstream bugs. The rule: every nullable column in a contract must document which meaning applies. Examples below.

The aggregation trap

SUM(column_with_nulls) ignores NULLs, but SUM(column_with_nulls) / COUNT(*) does not — COUNT(*) includes NULL rows. If you want the average over non-null rows, use AVG(col) or SUM(col) / COUNT(col). Mixing SUM and COUNT(*) is the single most common "the numbers are off" bug in dashboards that mix nullable measures.

The NULL-vs-zero revenue bug

Finance often wants "revenue = 0 for days with no sales." If your fact table uses NULL to mean "no sales," your MoM chart will produce NULL MoM for zero-revenue-days, not "down 100%." Generate a dense calendar grid (see Q4 in Part 09's SQL bank), COALESCE NULLs to 0, and you're safe.

↑ Back to contents

Streaming is where data engineering gets hard — not because the APIs are complex, but because the physics of distributed time makes correctness subtle. This file goes into watermark math, window mechanics, Flink's checkpoint algorithm, Kafka's internal protocols, and the proofs behind exactly-once semantics.

1. Unbounded Data — What Actually Changes

An unbounded source has four properties that break batch intuitions:

1. No sealed state. You never "have all the data." Every computation is "as of now," subject to revision.
2. Arbitrary lateness. Events can arrive minutes, hours, or days after they happened (mobile reconnects, buffering clients).
3. Out-of-order arrival. An event timestamped 10:05 may arrive after an event timestamped 10:07, because it took a different route through the system.
4. Continuous state. The processor must maintain state (windows, joins, aggregations) that grows without bound unless explicitly expired.

Consequence 1: "correct" becomes "correct as of watermark W"

A batch SQL SELECT SUM(watch_ms) GROUP BY dt has one answer. A streaming equivalent has an answer that keeps updating as late data arrives — and at any moment, the answer is "the sum of all data observed up to the current watermark, with the agreement that events older than W are considered closed."

Consequence 2: computations have four dimensions, not one

Beam's influential model:

• What are you computing? (the aggregation function)
• Where in event time? (the windowing scheme)
• When in processing time? (the trigger — when to emit a result)
• How do refinements relate? (accumulating, discarding, accumulating-with-retractions)

Naming them explicitly clarifies every streaming design decision.

Consequence 3: state is load-bearing

A batch job's state is transient (shuffle files, intermediate RDDs). A streaming job's state is the product. Lose it, lose correctness. Managing state durably (checkpoints, savepoints, RocksDB), bounding its growth (TTL, key expiration), and recovering it on failure (from checkpoint) dominates streaming engineering.

Consequence 4: time is a first-class citizen

In batch, "what time is it?" is a parameter. In streaming, time is the dataflow itself. Watermarks advance; triggers fire; windows close; state expires. Every operation is time-indexed.

2. The Three Times (Event, Processing, Ingestion)

Define precisely:

• Event time (ET) — when the event actually occurred in the real world. Stamped by the producer (the client, the sensor, the database transaction). This is what users care about. "Watch hours on 2026-04-18" means "hours of watching whose event_time falls on 2026-04-18."
• Ingestion time (IT) — when the event was received by the messaging system (e.g., the time Kafka assigned on append).
• Processing time (PT) — when the stream processor is evaluating the event.

Relationships:

• ET ≤ IT ≤ PT always (events can't be received before they happened; processing can't finish before receipt).
• IT - ET is the producer lateness — how long the event took to reach the messaging system.
• PT - IT is the processing lag — how behind the consumer is.
• PT - ET is the end-to-end latency experienced by consumers.

Distribution of ET → IT lag

Real production distributions are heavily right-skewed:

• p50: milliseconds to seconds.
• p99: tens of seconds to minutes.
• p99.9: hours (mobile reconnects, app-in-background).
• p99.99: days (offline clients syncing weeks later).

Lesson: watermarks must be tuned to the distribution, not the average. Setting a 1-minute lateness tolerance drops ~1% of events in most mobile datasets.

When does processing time ever make sense?

• Ad-hoc system health ("events/sec arriving right now").
• Simple at-least-once transforms with no time semantics.
• Systems that explicitly want "wall clock" behavior (alarms, heartbeats).

For any business metric, event time is correct. Always start with event time.

3. Watermarks — Math and Mechanics

A watermark is an assertion: "I assert that no more events with event_time ≤ W will arrive." The stream processor uses this to decide when windows can close.

Perfect vs heuristic watermarks

• Perfect watermark: actually correct. Only achievable when you have metadata (e.g., Kafka producer timestamps with bounded clock skew and source-aware committed offsets). Rare.
• Heuristic watermark: a guess, usually max(event_time_seen_so_far) - safety_margin. Wrong sometimes; that's what late-data handling is for.

Bounded out-of-orderness watermark

The common practical watermark:

W(t) = max(event_time seen before t) - B

where B is the bounded out-of-orderness (expected max lateness).

Example with B = 5 seconds:

• Stream sees events {ET: 10:00:00, ET: 10:00:02, ET: 10:00:01, ET: 10:00:04, ET: 10:00:03, ...}.
• After ET: 10:00:04 is observed, watermark = 10:00:04 - 5s = 09:59:59.
• Event ET: 10:00:03 arrives next: falls below watermark? 10:00:03 > 09:59:59, so not late. Accepted into windows.
• After ET: 10:00:10 is observed, watermark = 10:00:05. Events arriving with ET < 10:00:05 are now late.

Choosing B (bounded out-of-orderness)

Empirical approach:

1. Instrument production: log processing_time - event_time for every event.
2. Compute percentiles: p50, p95, p99, p99.9 of lateness.
3. Pick B as p99 or p99.9 — the tail you're willing to drop vs block.

Math: if B = p99 lateness, ~1% of events are late (dropped or routed to side output).

Higher B:

• Fewer late events.
• Longer window emission delays (latency ↑).
• Larger active state (windows held open longer).

Lower B:

• Shorter latency.
• More late events (correctness ↓).
• Smaller state.

Typical values: web/mobile analytics B = 30s–5m; sensor/IoT with good clocks B = 1–10s; batch-like (daily S3 drops) B = hours.

Watermark propagation in a dataflow graph

In Flink, each operator has its own watermark. Upstream operators propagate watermarks downstream. When an operator has multiple inputs, it takes the minimum watermark of its inputs:

source A: watermark advancing at 10:05
source B: watermark stuck at 10:02 (slow source)
joined operator: watermark = min(10:05, 10:02) = 10:02

This is correct: the downstream operator can't claim "no events before 10:05" because source B might still emit events with event time between 10:02 and 10:05.

Practical implication: a single slow source stalls the entire pipeline. Diagnose by watching per-operator watermarks in Flink UI. Solutions:

• Ensure all sources produce events regularly (heartbeat events if idle).
• Use WatermarkStrategy.forMonotonousTimestamps() where applicable (no out-of-orderness expected).
• Configure idle source detection — a source is marked idle after inactivity and its watermark contribution is ignored.

WatermarkStrategy<Event> strategy = WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withIdleness(Duration.ofMinutes(1))
    .withTimestampAssigner((ev, ts) -> ev.eventTime);

Watermarks across Kafka partitions

A single Kafka topic has multiple partitions. Each is an independent log. Event times within a partition are monotonic only if the producer writes in event-time order (usually not).

Flink's Kafka source computes per-partition watermarks, then combines via min. This is correct but introduces subtle issues:

• A partition with slow producers drags the watermark.
• An empty partition (no events) can cause the overall watermark to freeze unless idleness is configured.

Set watermarks per partition, with idleness detection, and ensure your partitioning key doesn't create empty partitions for prolonged times.

The "punctuated" variant

Instead of heuristics, the producer emits explicit watermark marker records: "I'm at event time T, no more events before T from me." Flink supports this via WatermarkGenerator with onEvent advancing and onPeriodicEmit emitting. Ideal when the source is a database CDC stream with explicit sequencing.

4. Windows — Types, State, and Trade-offs

A window groups events by time (or session). Each window maintains state (the aggregating values) and emits a result when it's ready.

Tumbling window

Fixed size, non-overlapping. Each event belongs to exactly one window.

|----W1----|----W2----|----W3----|
       ^           ^           ^
       events fall in one window

State: O(1) per key per active window. Simplest.

# PySpark Structured Streaming — 1-minute tumbling window
(events
    .withWatermark("event_ts", "5 seconds")
    .groupBy(F.window("event_ts", "1 minute"), "title_id")
    .agg(F.sum("watch_ms").alias("watch_ms")))

// Flink
events
    .keyBy(e -> e.titleId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new SumAggregator());

Sliding (hopping) window

Fixed size, slides forward by a step smaller than the size. Each event belongs to size/slide windows.

size=10m, slide=1m: every event is in 10 overlapping windows.

State: O(size/slide) × keys. Quickly becomes expensive.

Use when: you need rolling metrics ("watch hours in the trailing 10 minutes, updated every minute").

Tip: sliding windows can often be emulated by a tumbling window at the fine granularity + downstream rollup, which is cheaper.

Session window

Dynamic size, closes when there's a gap of inactivity longer than the session timeout.

events:  * * * *      * *     *    (gap >= timeout means new session)
windows: |------|     |---|   |-|

State: O(events per active session) × active sessions. Can be unbounded for users with continuous activity.

events
    .keyBy(e -> e.userId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
    .aggregate(new SessionBuilder());

Challenge: a single "always-active" user (a bot, a scraper, a stuck client) holds a session open forever, preventing emission. Solutions:

• Maximum session length (cap at e.g. 4 hours).
• Bot filtering at ingestion.
• Late "ejection" if max latency exceeded.

Global window

One window for the entire stream, with custom triggers deciding when to emit. Rarely correct unless you know exactly why you want it (e.g., a running counter that never resets).

Custom windows (e.g., calendar-aligned)

Windows aligned to calendar boundaries ("this day in user's local timezone", "this business week"). Implement with custom window assigners in Flink:

public class CalendarDayWindowAssigner extends WindowAssigner<Event, TimeWindow> {
    @Override
    public Collection<TimeWindow> assignWindows(Event ev, long ts, WindowAssignerContext ctx) {
        ZoneId zone = ZoneId.of(ev.userTimezone);
        ZonedDateTime zdt = Instant.ofEpochMilli(ev.eventTime).atZone(zone);
        ZonedDateTime dayStart = zdt.toLocalDate().atStartOfDay(zone);
        ZonedDateTime dayEnd   = dayStart.plusDays(1);
        return Collections.singleton(new TimeWindow(
            dayStart.toInstant().toEpochMilli(),
            dayEnd.toInstant().toEpochMilli()
        ));
    }
    // ... getDefaultTrigger, getWindowSerializer, etc.
}

Window state size

For N_keys keys and W_active simultaneously active windows:

• Tumbling: W_active = 1; state = O(N_keys).
• Sliding (size S, slide St): W_active = S/St; state = O(N_keys × S/St).
• Session: W_active = active_sessions; state = O(active_sessions × avg_events_per_session).

For high cardinality + long windows + many sliding steps: state explodes. Mitigations:

• RocksDB backend (disk-backed state, not memory-bound).
• Pre-aggregate (use aggregate not apply — maintains incremental state, not buffered events).
• TTL on state (Flink's StateTtlConfig).

5. Triggers and Emission Policy

A trigger decides when a window's current result is emitted. Default: on watermark (once per window, when watermark passes end-of-window). But you can do much more.

When to use custom triggers

• Speculative emission: emit partial results early, even before the window closes. Trade accuracy for latency.
• Per-row triggering: emit every N events (useful for low-latency monitoring).
• Processing-time-based: emit every 10 seconds regardless of watermark (for real-time dashboards).
• Data-driven: emit when some condition in the data is met (e.g., "emit when window's count exceeds threshold").

Example: early + on-watermark + late

events
    .keyBy(e -> e.titleId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .trigger(
        EventTimeTrigger.create()
            // speculative: emit every 10s while window is open (processing time)
            // Flink's built-in trigger doesn't combine these directly;
            // implement a custom Trigger extending EventTimeTrigger
    )
    .aggregate(new ViewerCountAgg());

Accumulation mode

When you emit multiple times for the same window, what do the downstream consumers see?

• Accumulating: each emission is the latest running total. Downstream must be an upsert sink that replaces the previous value. Example: (title_id=X, window=[10:00,10:01), count=42) replaces the earlier count=37.
• Discarding: each emission is only the new events since last emission. Downstream must sum across emissions to get the total. Harder downstream, but smaller messages.
• Accumulating + retracting: each emission includes a retraction of the previous value plus the new value. Allows downstream to handle both "live" updates and "final" snapshots. Most expressive, most expensive.

Choose based on downstream semantics. If sink is Kafka with compacted topic keyed by (title_id, window_start): accumulating + upsert. If sink is event log: retracting. If sink is an idempotent append-only store: discarding.

6. Late Data Strategies

An event is "late" if it arrives after its window's watermark has passed.

Strategy 1: Drop (default in Spark)

Late events are silently discarded. Fast, simple, incorrect for audit-sensitive workloads.

Strategy 2: Allowed lateness (extend window life)

Hold the window open past watermark by some duration; late events that arrive within this grace period update the window and trigger re-emission.

events
    .keyBy(e -> e.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.minutes(10))     // 10 extra minutes
    .aggregate(new Counter());

State cost: windows held for window_size + allowed_lateness. For a 1-minute window with 10-minute lateness, state is held 11 minutes per window.

Downstream must handle re-emissions — sink must be upsert-capable.

Strategy 3: Side output (recommended)

Route late events to a separate "late" stream for reconciliation downstream.

OutputTag<Event> lateTag = new OutputTag<>("late"){};

SingleOutputStreamOperator<Result> result = events
    .keyBy(e -> e.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .sideOutputLateData(lateTag)
    .aggregate(new Counter());

DataStream<Event> lateEvents = result.getSideOutput(lateTag);
lateEvents.addSink(new LateEventSink());  // write to separate topic/table

Dual-path reconciliation:

• Streaming pipeline handles on-time events.
• A batch job reads the late-event table daily and re-processes affected windows.
• Gold tables are rebuilt from the reconciled silver.

This is the production pattern. It's a Kappa + audit architecture.

Strategy 4: Retractions

Emit a retraction for the previous result and a new result reflecting the late event. Downstream must support retractions (a log of changes, not a snapshot).

Supported natively in Flink Table API with Retract mode. Upsert sinks handle retractions by deletion+insertion.

Measuring lateness in production

Emit a metric per event: lateness_ms = processing_time - event_time. Plot the distribution. Use to tune watermark B.

# Spark Structured Streaming: compute lateness column
events_with_lateness = events.withColumn(
    "lateness_ms",
    F.unix_millis(F.current_timestamp()) - F.unix_millis(F.col("event_ts"))
)

7. Delivery Semantics Proved

"Exactly-once" is often said without precision. Let's nail it down.

Definitions

Let M be a message in a stream. Let effect(M) be the externally-observable effect of processing M (a row written, a counter incremented, a record published).

• At-most-once: for each M, effect(M) happens 0 or 1 times.
• At-least-once: for each M, effect(M) happens 1 or more times.
• Exactly-once: for each M, effect(M) happens exactly 1 time.

The impossibility footnote

In a distributed system with possible crashes, the physical delivery of a message cannot be guaranteed exactly-once: any single network request may or may not have arrived despite no response. But the effect can be, by coordinating retries with deduplication or idempotent sinks.

Hence the industry term "effectively exactly-once": the physical operation may happen multiple times, but the side effect happens exactly once.

Three necessary conditions

For effectively-exactly-once:

1. Replayable source. On restart after failure, the system can re-consume from a precisely defined offset. (Kafka offsets satisfy this; naive HTTP push sources do not.)
2. Deterministic or idempotent processing. Re-processing the same input produces the same output, OR the sink absorbs duplicates idempotently.
3. Atomic commit coordinating source offsets and sink output. Either both the new offsets and the new output are committed, or neither.

Any one of these missing, and "exactly-once" fails.

Spark Structured Streaming's exactly-once

Spark provides effectively-exactly-once for specific combinations:

Mechanism:

1. Source tracks offsets per micro-batch (stored in offsets/ in checkpoint directory).
2. Each micro-batch's computation is deterministic given the offsets.
3. Sink commits atomically:
   • File sinks (Parquet, Iceberg, Delta): use a two-phase commit with a commits/ log. After the data files are written, a commit marker is appended atomically. On restart, uncommitted micro-batches are re-executed.
   • Kafka sink: uses Kafka transactions (transactional.id). Micro-batch produces within a transaction; transaction commits with the offset commit.
4. On failure, restart reads offsets/ and commits/ to find the last committed micro-batch, re-executes the next one deterministically.

Where it breaks:

• Non-deterministic query (rand(), current_timestamp(), reading from a mutable source).
• Non-atomic sinks (HTTP POST, naive JDBC without dedup key).
• Kafka transactions disabled or misconfigured.
• Checkpoint directory corruption.

(stream
    .writeStream
    .format("iceberg")
    .option("checkpointLocation", "s3://ckpt/events/")
    .trigger(processingTime="30 seconds")
    .start())

Spark guarantees EO when using checkpoints + Iceberg/Delta sinks + deterministic transforms.

Flink's exactly-once via 2PC and checkpoints

Flink uses the Chandy-Lamport algorithm (distributed snapshot) for checkpoints, combined with two-phase commit at sinks.

Barrier-based checkpointing:

1. JobManager injects a barrier with checkpoint ID C into each source partition.
2. As barriers flow through the dataflow graph, each operator aligns: when it has received the barrier from all its inputs, it snapshots its state to durable storage.
3. State snapshot includes: operator state, keyed state, source offsets.
4. Operator forwards the barrier to all its outputs.
5. Sinks acknowledge completion. When all sinks ack, checkpoint C is globally complete.

[Source] ──barrier──▶ [Map] ──barrier──▶ [Window] ──barrier──▶ [Sink]
    ↓                    ↓                  ↓                    ↓
 snapshot             snapshot           snapshot             commit

Aligned vs unaligned barriers:

• Aligned: operator waits for barriers from all inputs before snapshotting. Clean semantics but blocks processing of inputs whose barrier arrived first.
• Unaligned (Flink 1.11+): operator snapshots immediately when first barrier arrives, including in-flight records. Less backpressure sensitivity; larger snapshots. Enabled via env.getCheckpointConfig().enableUnalignedCheckpoints().

Two-phase commit at sinks:

1. On barrier: sink does a pre-commit (write to staging, reserve transaction ID).
2. When JobManager confirms the checkpoint globally complete: sink does commit (make data visible).
3. On restart: replay uncommitted pre-commits as needed.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60_000);  // every 60s
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);
env.getCheckpointConfig().setCheckpointTimeout(600_000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

What EO does NOT guarantee

• Ordering across keys: events may still be emitted to the sink in an order different from their arrival.
• No retractions: updates to already-committed results are still visible downstream as new records.
• Cross-system atomicity: EO within one Flink job + one Kafka cluster. Across two Kafka clusters, or Kafka + a different sink, EO requires explicit 2PC support end-to-end.
• Effective-exactly-once for the system, not the business: if the input data is duplicated (upstream producer retried without dedup), EO still processes both copies.

Idempotent sink as an alternative to 2PC

Often simpler than 2PC is to make the sink idempotent by key:

• Write each record with a unique idempotency key.
• The sink's INSERT IGNORE / ON CONFLICT DO NOTHING / upsert-by-key handles duplicates.
• Combined with at-least-once semantics, this gives effective-exactly-once.

-- Postgres sink with idempotency
INSERT INTO sink_table (event_id, user_id, ...)
VALUES (?, ?, ...)
ON CONFLICT (event_id) DO NOTHING;

Works well; doesn't require sink-coordinator awareness. Limit: sinks that can't enforce uniqueness (append-only files) don't support this pattern.

8. Flink Internals — Dataflow, Barriers, State

Architecture

• JobManager: coordinator; runs the ExecutionGraph, schedules tasks, coordinates checkpoints.
• TaskManagers: workers; each runs N slots, each slot runs a subtask of an operator.
• ResourceManager: allocates TaskManagers (typically YARN or Kubernetes).
• Dispatcher: receives jobs, spawns JobManagers per job.

Tasks, subtasks, and chains

A Flink job is a DAG of operators. Each operator has a parallelism (how many parallel instances run). Each instance is a subtask. A JobManager assigns subtasks to TaskManager slots.

Operator chaining: if an operator's output goes to another operator 1:1 and no shuffle is needed, Flink chains them into a single thread. Reduces serialization overhead hugely. forward channels are chained; rebalance, keyBy, etc. are not.

DataStream<Event> events = env.addSource(kafkaSource);         // source
DataStream<Event> filtered = events.filter(ev -> ev.valid);    // chained with source
KeyedStream<Event, String> keyed = filtered.keyBy(ev -> ev.userId);  // NOT chained (keyBy shuffles)
DataStream<Result> windowed = keyed.window(...).aggregate(...); // new chain

Task chain = single thread. Processes records with zero network hops and zero ser/de. Fast.

Keyed state vs operator state

• Keyed state: one value per key per operator. keyBy partitions the stream; each partition holds state for its keys. The main way Flink scales stateful computation.
  • ValueState<T>, ListState<T>, MapState<K,V>, ReducingState<T>, AggregatingState<IN,OUT>.
• Operator state: per-subtask state, not per-key. Used by sources (offset tracking) and custom operators. Sub-types: broadcast state, list state (union or split).

public class CountByUserFn extends KeyedProcessFunction<String, Event, Result> {
    private ValueState<Long> count;

    @Override
    public void open(Configuration cfg) {
        ValueStateDescriptor<Long> d = new ValueStateDescriptor<>("count", Long.class, 0L);
        d.enableTimeToLive(StateTtlConfig.newBuilder(Time.hours(24)).build());
        count = getRuntimeContext().getState(d);
    }

    @Override
    public void processElement(Event ev, Context ctx, Collector<Result> out) throws Exception {
        Long c = count.value();
        c += 1;
        count.update(c);
        out.collect(new Result(ctx.getCurrentKey(), c));
    }
}

State backends

• HashMapStateBackend: in-JVM-memory. Fastest. State bounded by heap. Synchronous checkpoints (pauses JVM). Use for small state (< ~2GB).
• EmbeddedRocksDBStateBackend: on-disk key-value store (RocksDB). State bounded by local disk. Asynchronous and incremental checkpoints. Use for state > a few GB.

RocksDB internals:

• SSTables (Sorted String Tables) on disk, immutable.
• Memtable in memory; flushed to SSTable periodically.
• LSM tree structure: multiple levels of SSTables, compacted periodically.
• State read/write: serialize key, look up in memtable + levels (with bloom filters).

Incremental checkpoints

With RocksDB, Flink checkpoints only the SSTables that changed since the last checkpoint. New SSTables uploaded; unchanged ones referenced by manifest. Reduces checkpoint time from "full state" to "delta state."

checkpoint N-1: [sst1, sst2, sst3]
operator writes, compaction happens
checkpoint N:   [sst1 (ref), sst4, sst5]  <-- only sst4 and sst5 uploaded

Savepoints vs checkpoints

• Checkpoints: system-owned, periodic, optimized for fast recovery. May be incremental.
• Savepoints: user-triggered, versioned, portable across Flink versions (with caveats). Use for explicit restarts (deploys, cluster migrations, schema changes).

flink savepoint <job_id> s3://savepoints/myjob/

flink run -s s3://savepoints/myjob/savepoint-xxx <new-jar>.jar

State migration

When you change the shape of state (add a field, change types), you need a migration. Flink supports:

• Schema evolution via Avro-serialized state (auto).
• Explicit state migration: write a one-off job that reads the savepoint with old schema, writes with new.
• Keyed state type migration: limited; usually requires re-keying from scratch.

9. Kafka Internals — ISR, Controllers, Exactly-Once

Topic, partition, offset

• Topic: named append-only log, split into partitions for parallelism.
• Partition: an ordered sequence of records; each record has a monotonically increasing offset within the partition.
• Producers choose partition (via key hash or explicit selection).
• Consumers read partitions sequentially; track their offset.

Parallelism = number of partitions. One consumer per partition per consumer group is the maximum (more consumers than partitions → idle consumers).

Replication

Each partition has N replicas (typically 3):

• Leader: handles reads and writes.
• Followers: pull from the leader asynchronously.
• ISR (In-Sync Replicas): the set of followers that are within replica.lag.time.max.ms of the leader.

Writes:

• acks=0: fire-and-forget. Producer doesn't wait for leader ack.
• acks=1: wait for leader to persist. Lost on leader failure before followers catch up.
• acks=all: wait for all ISRs to persist. No loss within replication factor (assuming min.insync.replicas >= 2).

Controller and leader election

One broker is the controller (elected via ZooKeeper or KRaft). Controller maintains cluster metadata — which broker leads which partition.

On broker failure:

1. Controller detects failure (ZooKeeper session expiry or KRaft heartbeat timeout).
2. Controller removes failed broker from ISR for partitions it led.
3. Controller elects a new leader from surviving ISR (first in the ISR list).
4. Controller propagates leadership change to all brokers and producers.

Without unclean leader election, the new leader is guaranteed to have every committed record. With unclean leader election enabled (unclean.leader.election.enable=true), a non-ISR replica can be elected — at the cost of data loss.

KRaft (ZooKeeper-free Kafka)

Kafka 3.x+ supports KRaft mode: the controller runs internally via Raft. No ZooKeeper. Simpler operationally; faster metadata propagation; supports larger clusters (millions of partitions).

Default in new deployments from 2024 onward.

Kafka exactly-once

Kafka offers exactly-once semantics (EOS) for specific patterns:

Producer idempotence (enable.idempotence=true):

• Producer is assigned a unique producer_id (PID) by the broker.
• Each record has a sequence number per partition.
• Brokers deduplicate based on (PID, partition, sequence) for the lifetime of the PID's epoch.
• Prevents duplicates from producer retries within a single producer session.

Transactional producers (transactional.id=my-tx):

• Producer starts a transaction, writes to multiple partitions/topics, commits.
• All writes are atomic — consumers with isolation.level=read_committed see either all or none.
• Critical for exactly-once stream processing where "read from topic A, write to topics B and C" must be all-or-nothing.

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("out_topic_1", ...));
    producer.send(new ProducerRecord<>("out_topic_2", ...));
    producer.sendOffsetsToTransaction(consumedOffsets, consumerGroup);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
    throw e;
}

Consumer rebalance protocol

When consumers in a group join or leave:

Eager rebalance (classic):

1. Coordinator triggers rebalance.
2. All consumers revoke their partitions (stop processing entirely).
3. Coordinator assigns partitions via strategy.
4. Consumers resume.

Stop-the-world can be minutes with many consumers / many partitions. Kills latency SLOs.

Cooperative sticky rebalance (Kafka 2.4+):

1. Coordinator identifies partitions to move.
2. Only affected consumers revoke only the moving partitions.
3. New assignments go to target consumers.
4. Non-affected consumers continue uninterrupted.

Minimal disruption. Prefer in all modern deployments: partition.assignment.strategy=CooperativeStickyAssignor.

Log compaction

For "changelog" topics keyed by entity ID, Kafka can compact:

• Keeps only the latest record per key.
• Older records with the same key are garbage-collected during compaction.
• Tombstones (null-valued records) delete the key permanently after a grace period.

log before compaction:
  key=user1, val={name:A}
  key=user2, val={name:X}
  key=user1, val={name:A'}
  key=user1, val=null          (tombstone)
  key=user2, val={name:Y}

log after compaction:
  key=user1, val=null          (to be deleted after grace)
  key=user2, val={name:Y}

Compacted topics serve as durable key-value stores readable as streams. Used for:

• Materializing database CDC (final state of each row).
• State topic backing Kafka Streams applications.
• Configuration distribution.

10. Streaming Joins

Joining two unbounded streams is the hardest operation in streaming. There are several flavors.

Stream-stream window join

Join two streams within a time window: match pairs (a, b) where a.timestamp and b.timestamp fall within a common window and the join key matches.

clicks.join(impressions)
    .where(c -> c.userId).equalTo(i -> i.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .apply(new JoinFn());

State: O(|clicks| + |impressions|) within the window. Grows linearly with window size and rate.

Stream-stream interval join

Match event pairs within a relative time interval: for each a, find b where a.ts - X ≤ b.ts ≤ a.ts + Y.

clicks.keyBy(c -> c.userId)
    .intervalJoin(impressions.keyBy(i -> i.userId))
    .between(Time.minutes(-1), Time.minutes(5))
    .process(new IntervalJoinFn());

Used for causal analysis (impressions preceding clicks) where window alignment would be wrong.

Stream-table (temporal) join

Join a stream of events against a slowly-changing table (often a dim broadcast or a compacted topic). Each event looks up the version of the table that was current at event time.

Implementation patterns:

• Broadcast state: broadcast the table updates to all parallel instances of the join operator. Each instance keeps the table in local state. Great when the table is small (< 1 GB).
• Keyed state + async lookup: each event does an async DB lookup, keeping a cache in keyed state.
• Versioned table joins (Flink Table API): FOR SYSTEM_TIME AS OF syntax.

-- Flink SQL: temporal join
SELECT o.*, c.country, c.subscription_tier
FROM orders o
JOIN user_dim FOR SYSTEM_TIME AS OF o.event_ts AS c
  ON o.user_id = c.user_id;

The engine maintains the history of user_dim; the join picks the version of each user that was current at event_ts.

Stream enrichment via async I/O

When the dimension is too large for broadcast state but must be looked up per event:

AsyncDataStream.unorderedWait(
    events,
    new AsyncDatabaseLookup(),
    30, TimeUnit.SECONDS,     // timeout
    1000                       // capacity (max concurrent lookups)
).map(enriched -> process(enriched));

Concurrent async lookups keep the pipeline fed while external systems respond. Critical: use unorderedWait if order doesn't matter (higher throughput); orderedWait otherwise.

11. State Management at Scale

Production streaming pipelines accumulate TBs of state. Managing it is its own discipline.

State TTL

Per-state configurable time-to-live. Flink's StateTtlConfig:

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.hours(24))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter(1000)
    .build();

Cleanup strategies:

• In-access: expired state is removed on read (cheap but delays cleanup).
• Full-snapshot: cleanup during checkpoint (exhaustive but costly).
• RocksDB compaction filter: cleanup during RocksDB's background compaction. Preferred for large state.

Bounded session windows via custom expiration

Session windows without explicit max length can grow without bound. Implement max session length:

public class BoundedSessionFn extends KeyedProcessFunction<String, Event, Session> {
    private ValueState<Session> current;
    private ValueState<Long> expirationTimer;

    @Override
    public void processElement(Event ev, Context ctx, Collector<Session> out) throws Exception {
        Session s = current.value();
        if (s == null || ev.ts > s.lastEventTs + SESSION_GAP_MS) {
            // new session
            if (s != null) out.collect(s);
            s = new Session(ev);
        } else {
            s.add(ev);
            if (s.durationMs() > MAX_SESSION_MS) {
                out.collect(s);
                s = new Session(ev);   // start a new session from this event
            }
        }
        current.update(s);
        ctx.timerService().registerEventTimeTimer(ev.ts + SESSION_GAP_MS);
    }

    @Override
    public void onTimer(long ts, OnTimerContext ctx, Collector<Session> out) throws Exception {
        Session s = current.value();
        if (s != null && ts >= s.lastEventTs + SESSION_GAP_MS) {
            out.collect(s);
            current.clear();
        }
    }
}

State partitioning evolution

Change the key you're partitioning on? Flink doesn't let you; you must:

1. Savepoint current state.
2. Write a migration job that reads savepoint, rewrites state under new keying.
3. Start new job from migrated savepoint.

Or: start fresh, lose history. Painful; plan keys carefully upfront.

State size budgeting

Estimate:

state_size = num_keys × bytes_per_key_state × num_windows_active_per_key

Example:

• 100M users
• 1 KB state per user (a few counters, last-event-ts)
• 1 window active per user (current session)
• → 100 GB of state.

RocksDB backend + local SSD + incremental checkpoints handle this fine. HashMap backend wouldn't (100 GB heap is infeasible).

For 1B users × 1 KB = 1 TB state: still viable with RocksDB + fast SSDs. Above that, consider sharding across multiple Flink jobs or moving state to an external store with async lookup.

12. Backpressure and Flow Control

Backpressure: when a downstream operator can't keep up, upstream must slow down. If it doesn't, unbounded buffers, OOM, or data loss follow.

How Flink signals backpressure

Flink uses credit-based flow control. Each downstream operator advertises how much buffer space it has (credit); upstream only sends when credit is available.

If downstream is slow, credit is consumed faster than replenished, and upstream naturally blocks. This propagates back to the source, which (if Kafka) simply stops pulling records.

Diagnosing backpressure

In the Flink UI, each operator shows backpressure status (OK / LOW / HIGH). Trace the chain:

• Find the operator showing HIGH backpressure.
• Its downstream is the bottleneck.

Metrics to check:

• outputQueueLength: how full the output buffer is.
• inputQueueUsage: same for inputs.
• numRecordsInPerSecond vs numRecordsOutPerSecond: disparity indicates buffering.

Fixing backpressure

• Increase downstream parallelism: run more instances of the slow operator.
• Optimize the slow operator: profile; CPU? I/O? state access? The usual culprits are: Python UDFs (in PyFlink), synchronous external calls, state backend misuse.
• Use async I/O for external lookups.
• Shard state across more keys: if one key is hot, repartition with a salt (see Spark's skew fix — same principle applies).

Backpressure and checkpoints

Aligned checkpoints require barriers to reach operators; if an operator is backpressured, barriers pile up behind records in the input buffer, delaying the snapshot. Severe backpressure → checkpoint timeouts → job restarts → fallback loop.

Unaligned checkpoints (Flink 1.11+) mitigate: barriers "jump" ahead of in-flight records in the buffer. The snapshot includes those records as part of the channel state. Larger snapshots, but doesn't stall under backpressure.

env.getCheckpointConfig().enableUnalignedCheckpoints();

Enable in production where backpressure is possible.

13. Lambda vs Kappa Revisited

Lambda architecture

Two pipelines:

• Batch layer — periodic, fully-accurate, high-latency.
• Speed layer — streaming, best-effort, low-latency.
• Serving layer — merges batch and speed views; clients see the combined result.

Problems:

• Two codebases. Every business rule implemented twice. Drift inevitable.
• Two debugging surfaces. Which layer produced this wrong number?
• Two SLA sets. When one layer lags, is the combined answer correct?

Kappa architecture

One pipeline, streaming. Batch is "streaming with a bigger window" or "streaming over a replay."