Grain = "what does one row mean?" Define explicitly per fact table.
Slowly Changing Dimensions
Type
Behavior
Type 0
No change tracking; ignore updates
Type 1
Overwrite — current truth only
Type 2
New row per change with valid_from / valid_to + is_current
Type 3
Add prior_value column
Type 4
Mini history table
Type 6
Hybrid 1+2+3 (current + history + previous)
SCD Type 2: one row per version, validity intervals
SCD Type 2 via MERGE
SCD2 in one statement is a two-pass idea expressed in a single MERGE: close the current row when an attribute changed, then insert the new version. The trick is the duplicate source rows — one to match-and-expire, one (unmatched) to insert as the new current row.
-- ANSI MERGE (Snowflake / BigQuery / Spark-Iceberg dialects vary slightly).
-- `src` = today's snapshot of the dimension's natural key + attributes.
MERGE INTO dim_customer AS t
USING (
-- row A: closes a changed current row; also inserts a brand-new key
-- (no current row to match -> falls through to NOT MATCHED)
SELECT customer_id AS join_key, customer_id, tier, region FROM src
UNION ALL
-- row B: NULL join_key never matches -> forces the INSERT of the new
-- version, but ONLY for EXISTING keys whose attributes changed
SELECT NULL AS join_key, src.customer_id, src.tier, src.region
FROM src
JOIN dim_customer c
ON c.customer_id = src.customer_id AND c.is_current = TRUE
WHERE src.tier <> c.tier OR src.region <> c.region
) AS s
ON t.customer_id = s.join_key
AND t.is_current = TRUE
WHEN MATCHED
AND (t.tier <> s.tier OR t.region <> s.region) -- attribute changed
THEN UPDATE SET t.is_current = FALSE,
t.valid_to = CURRENT_TIMESTAMP()
WHEN NOT MATCHED
THEN INSERT (customer_id, tier, region, valid_from, valid_to, is_current)
VALUES (s.customer_id, s.tier, s.region,
CURRENT_TIMESTAMP(), NULL, TRUE);
Silver (cleaned) — typed, dedup, joined with conformed dims. Useful but not business-shaped.
Gold (mart) — denormalized, business metrics, aggregates, fit-for-BI.
Reprocess Silver/Gold from Bronze; never edit Bronze.
Medallion flow: Bronze → Silver → Gold
One Big Table (OBT) vs Star
Modern columnar engines tolerate wide denorm tables. OBT = simpler queries, faster on dashboards; star = better for ad-hoc + smaller writes. Hybrid: star at Silver, OBT at Gold for hot dashboards.
MODULE 5
dbt & Transformation
SQL + Jinja + DAG. Tests + docs + lineage built in.
Core Concepts
Models — SELECT in .sql files; dbt wraps as table / view / incremental.
Transactional producer + idempotent + read-committed; or sink dedup
Idempotent sink: dedup by event_id
The pragmatic "effectively-once" pattern when you control the sink but not a full 2PC: make every write idempotent on a stable business key so replays collapse. With an open table format you express it as a keyed MERGE.
# PySpark Structured Streaming -> idempotent Iceberg upsert via foreachBatch.
# Replays re-deliver the same event_id; MERGE makes the write a no-op.
def upsert_batch(micro_df, batch_id):
# 1) collapse duplicates WITHIN the micro-batch (keep newest per key)
from pyspark.sql import functions as F, Window
w = Window.partitionBy("event_id").orderBy(F.col("event_ts").desc())
deduped = (micro_df
.withColumn("rn", F.row_number().over(w))
.filter(F.col("rn") == 1)
.drop("rn"))
deduped.createOrReplaceTempView("updates")
# 2) idempotent merge: insert new keys, ignore (or refresh) existing ones
micro_df.sparkSession.sql("""
MERGE INTO catalog.db.events AS t
USING updates AS s
ON t.event_id = s.event_id
WHEN NOT MATCHED THEN INSERT *
""")
(spark.readStream.format("kafka")
.option("subscribe", "events")
.load()
.selectExpr("CAST(value AS STRING) AS json")
# ... parse json -> event_id, event_ts, payload ...
.writeStream
.foreachBatch(upsert_batch)
.option("checkpointLocation", "s3://bucket/_chk/events") # offsets + state
.start())
Windows
Tumbling — fixed, non-overlapping (every 5 min).
Hopping / sliding — fixed size, slide step (5 min size, 1 min step).
Session — close after gap of inactivity (user activity sessions).
Event time vs processing time — event time tolerates late arrivals via watermarks.
Watermark — bound on "no event older than this is expected". Closes window.
# Event-time tumbling aggregation. withWatermark bounds how long
# late state is kept (10 min) AND defines the late-data drop threshold.
from pyspark.sql import functions as F
agg = (events
.withWatermark("event_ts", "10 minutes") # tolerate 10 min lateness
.groupBy(
F.window("event_ts", "5 minutes"), # 5-min tumbling window
"country")
.agg(F.count("*").alias("events"),
F.approx_count_distinct("user_id").alias("uniques")))
(agg.writeStream
.outputMode("update") # emit changed windows; 'append' emits only on close
.format("console")
.option("checkpointLocation", "s3://bucket/_chk/win")
.start())
The equivalent in Flink SQL uses a windowing TVF over a column with a declared watermark:
-- The source table declares the watermark in its DDL:
-- WATERMARK FOR event_ts AS event_ts - INTERVAL '10' MINUTE
SELECT
window_start, window_end, country,
COUNT(*) AS events,
COUNT(DISTINCT user_id) AS uniques
FROM TABLE(
TUMBLE(TABLE events, DESCRIPTOR(event_ts), INTERVAL '5' MINUTE))
GROUP BY window_start, window_end, country;
Stream Engines
Flink — true streaming, EOS, savepoints, stateful ops at scale. Industry standard for complex pipelines.
Row-level upsert is the headline feature of an open table format over plain Parquet. The MERGE grammar is the same on Iceberg (Spark/Trino) and Delta; only the maintenance command differs.
-- Upsert a CDC/staging batch into an Iceberg or Delta table.
MERGE INTO catalog.db.orders AS t
USING staging_orders AS s
ON t.order_id = s.order_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED AND s.op <> 'D' THEN INSERT *;
After many small commits, run compaction to fight the small-file problem:
-- Iceberg: rewrite small data files into ~512 MB targets (Spark procedure)
CALL catalog.system.rewrite_data_files(
table => 'db.orders',
options => map('target-file-size-bytes','536870912'));
-- Delta Lake equivalent
OPTIMIZE catalog.db.orders; -- bin-packs files; ZORDER BY (col) for clustering
MODULE 9
Data Quality & Contracts
Catch broken pipelines before consumers do.
Test Categories
Category
Examples
Schema
Required cols present; types match; nullability
Uniqueness / referential
PK unique; FK present in dim
Range / accepted values
0 ≤ rate ≤ 1; status ∈ {…}
Volume
Row count within ±X% of 7-day avg
Freshness
max(updated_at) within SLA
Distribution
Mean / null-rate within control limits
Custom business rule
Σ(line_items.amount) = orders.total
Tools
dbt tests — fast, in-pipeline. Unique / not_null / relationships built in.
Great Expectations — Python-driven expectations + docs.
Soda — declarative YAML checks.
Monte Carlo / Bigeye / Anomalo — ML-based anomaly + lineage.
Open standards: OpenLineage, OpenMetadata.
Data Contracts
Producer publishes typed schema + SLA to consumers.
Wide deps (groupBy, join, repartition) move data across nodes — expensive.
Minimize shuffles: filter + project early, broadcast small side of join (< 10 MB default), avoid groupBy when reduceByKey suffices.
Skew: hot key concentrates work on one task. Salt key, AQE skew join, or split.
Spill to disk when memory exceeded — ensure shuffle dirs sized + fast disks.
Shuffle: map → reduce partitions, with a skewed key
Skew join: AQE first, salting when AQE can't
Spark 3.0+ AQE auto-splits skewed partitions at runtime — try it before hand-rolling anything:
# AQE detects a partition that is both > skewedPartitionFactor x median
# AND > skewedPartitionThresholdInBytes, then splits it across tasks.
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
When AQE isn't enough (e.g. a single mega-key, or a non-sort-merge join), salt the hot side and replicate the small side so the work fans out:
from pyspark.sql import functions as F
N = 16 # salt buckets: split each hot key into N sub-keys
# 1) skewed (large) side: append a random salt 0..N-1 to the join key
fact_salted = fact.withColumn("salt", (F.rand() * N).cast("int"))
# 2) small side: replicate every row once per salt value (cross with 0..N-1)
salts = spark.range(N).withColumnRenamed("id", "salt")
dim_salted = dim.crossJoin(salts)
# 3) join on (key, salt) -> the hot key is now spread over N tasks
out = fact_salted.join(dim_salted, ["join_key", "salt"], "inner").drop("salt")
-- naive (leakage)
join features f on f.user_id = label.user_id
-- correct (point-in-time, AS OF label.event_ts)
join features f
on f.user_id = label.user_id
and f.valid_from <= label.event_ts
and (f.valid_to is null or f.valid_to > label.event_ts)
Embeddings as Data
DE owns the pipeline: text/image → embedding → indexed vector store. Tracking + versioning + reindex on model change. Integrates with RAG (see AI Engineer notebook).
MODULE 13
Cheat Sheet
Default stack + decision rules.
Default Stack
Storage: S3 + Iceberg
Warehouse: Snowflake / BigQuery / Databricks SQL
Ingest: Fivetran / Airbyte + Debezium CDC
Transform: dbt
Orchestrate: Airflow or Dagster
Stream: Kafka + Flink
Quality: dbt tests + GE / Soda
Catalog: DataHub / OpenMetadata
BI: Looker / Lightdash / Metabase
Pick Storage
Pure analytics + SQL → warehouse (Snowflake / BQ)
Multi-engine + ML + cheap → lakehouse (Iceberg + Spark)
OLTP-style updates → Postgres / Aurora; CDC out
K-V serving → DynamoDB / Cassandra
Full-text + vector → OpenSearch / Vespa
Streaming Defaults
Partition key = entity needing order
Avro + schema registry
At-least-once + idempotent sink
Watermark on event time
Compact topic for "current state"
Dead-letter queue for poison events
Modeling Defaults
Bronze (raw) → Silver (clean) → Gold (mart)
Star schema at Silver; OBT at Gold for hot dashboards
The whiteboard SQL, end-to-end design reps, and Airflow internals DE loops actually test.
Window Functions: The Interview Workhorse
A window function computes across a set of rows related to the current row without collapsing them (unlike GROUP BY). The OVER() clause defines the window: PARTITION BY resets the computation per group, ORDER BY sequences rows, and the frame clause bounds which rows feed the aggregate. Master five functions and you cover ~80% of SQL rounds.
ROW_NUMBER() — 1,2,3,4 with no ties. Deterministic only if ORDER BY is unique; add a tiebreaker column.
RANK() — 1,2,2,4: ties share a rank, then skip. Use for "top-N with ties kept".
DENSE_RANK() — 1,2,2,3: ties share, no gap. Use for "Nth distinct salary".
LAG(col, n) / LEAD(col, n) — value from n rows back/ahead in the partition. The tool for deltas, run-length, and gap detection.
SUM/COUNT ... OVER(...) — running totals and moving windows via a frame.
-- 2nd highest DISTINCT salary per department (handles ties correctly)
SELECT department_id, salary
FROM (
SELECT department_id, salary,
DENSE_RANK() OVER (PARTITION BY department_id
ORDER BY salary DESC) AS rnk
FROM employees
) t
WHERE rnk = 2;
-- 7-day trailing moving average of daily revenue
SELECT day, revenue,
AVG(revenue) OVER (ORDER BY day
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS ma7
FROM daily_rev;
Deduplication with QUALIFY
The canonical "keep the latest row per key" problem. Classic SQL forces a self-join or a subquery on ROW_NUMBER() because you cannot filter a window function in WHERE (windows are computed after WHERE). Snowflake, BigQuery, DuckDB, and Databricks add QUALIFY, which filters on window results the way HAVING filters on aggregates — one clean statement.
-- Latest event per user (QUALIFY dialects: Snowflake/BigQuery/DuckDB)
SELECT *
FROM events
QUALIFY ROW_NUMBER() OVER (PARTITION BY user_id
ORDER BY event_ts DESC, event_id DESC) = 1;
-- Portable equivalent (Postgres/MySQL 8+): wrap in a subquery
SELECT * FROM (
SELECT *,
ROW_NUMBER() OVER (PARTITION BY user_id
ORDER BY event_ts DESC, event_id DESC) AS rn
FROM events
) t
WHERE rn = 1;
Tiebreaker matters — two rows with the same event_ts make ROW_NUMBER nondeterministic. Append a monotonic key (event_id, ingestion sequence) to ORDER BY.
Full dedup (all columns identical) — cheaper to SELECT DISTINCT or GROUP BY every column; reserve window dedup for "latest per business key".
Gaps-and-Islands
"Find consecutive runs" (islands) and the holes between them (gaps): consecutive login days, uninterrupted subscription periods, contiguous free seat ranges. The trick is the difference of two sequences. Number rows with ROW_NUMBER(), subtract that from the value itself; rows in the same run share a constant difference, giving you a group key to aggregate on.
-- Longest streak of consecutive active days per user
WITH ranked AS (
SELECT user_id, activity_date,
ROW_NUMBER() OVER (PARTITION BY user_id
ORDER BY activity_date) AS rn
FROM (SELECT DISTINCT user_id, activity_date FROM activity) d
),
grouped AS (
SELECT user_id, activity_date,
-- island key: date minus its ordinal is constant within a run
activity_date - (rn * INTERVAL '1 day') AS grp
FROM ranked
)
SELECT user_id,
MIN(activity_date) AS streak_start,
MAX(activity_date) AS streak_end,
COUNT(*) AS streak_len
FROM grouped
GROUP BY user_id, grp
ORDER BY streak_len DESC;
Integer sequences — for numeric IDs use id - ROW_NUMBER() directly; no interval math.
De-dup first — two rows on the same day break the "one ordinal per step" invariant. The inner DISTINCT is not optional.
Gaps — same idea inverted: LEAD(date) minus current date > 1 marks a gap boundary.
Sessionization with LAG
Group an event stream into sessions where a new session starts after an inactivity gap (web analytics standard: 30 minutes). Compute the gap to the previous event with LAG, flag a boundary when the gap exceeds the threshold, then a running SUM of those flags becomes the session id.
-- 30-minute inactivity sessionization
WITH gaps AS (
SELECT user_id, event_ts,
LAG(event_ts) OVER (PARTITION BY user_id
ORDER BY event_ts) AS prev_ts
FROM events
),
flagged AS (
SELECT user_id, event_ts,
CASE WHEN prev_ts IS NULL
OR event_ts - prev_ts > INTERVAL '30 minutes'
THEN 1 ELSE 0 END AS is_new_session
FROM gaps
)
SELECT user_id, event_ts,
SUM(is_new_session) OVER (PARTITION BY user_id
ORDER BY event_ts) AS session_id
FROM flagged;
End-to-End Pipeline Walkthrough
Design an events pipeline: clickstream from a web app to dashboards and a churn model. Reason through the four stages — ingest → store → transform → serve — and name a technology and a failure mode at each hop.
Ingest — SDK POSTs events to a collector behind a load balancer, which produces to Kafka (topic partitioned by user_id for per-user ordering). Kafka is the durable buffer that decouples spiky producers from slower consumers; retention 7 days lets you replay.
Store (raw / Bronze) — a Kafka Connect / Flink sink writes Avro or Parquet to S3 as an Iceberg table, partitioned by event_date. Land raw and immutable; never transform on the write path so you can always reprocess.
Transform (Silver → Gold) — dbt on Spark/Snowflake models: Silver = deduped, typed, sessionized; Gold = business aggregates (DAU, funnel, features). Incremental models process only new partitions.
Serve — Gold tables feed BI (Looker/Tableau) for dashboards, a feature store (Feast) for the churn model, and reverse-ETL (Hightouch) back into the CRM. Orchestrated by Airflow; quality gates by dbt tests + Great Expectations.
Idempotency spine — every event carries an event_id (UUID) + event_ts. Exactly-once is approximated as at-least-once delivery + dedup on event_id in Silver.
Schema contract — a schema registry (Avro/Protobuf) rejects incompatible producer changes before they poison the lake.
Late data — partition by event time, not processing time; reprocess a partition when late events arrive (watermark or lookback window).
Back-of-Envelope Capacity & Cost
Interviewers want a defensible number in 60 seconds, not precision. Anchor on events/day → bytes → $/month. Assume 10M daily active users, 20 events each = 200M events/day, ~1 KB raw JSON per event.
Quantity
Estimate
How
Events/day
200M
10M DAU × 20
Avg throughput
~2,300 events/s
200M / 86,400 s
Peak throughput
~11,500 events/s
5× diurnal peak factor
Raw ingest/day
~200 GB
200M × 1 KB
Parquet on disk
~20 GB/day
~10× compression
Yearly stored
~7.3 TB
20 GB × 365
S3 storage cost
~$2,015/yr
7.3 TB × $0.023/GB-mo × 12
Kafka partitions
~46
peak 11.5k/s ÷ ~500/s per partition, ×2 headroom
Peak factor — real traffic is spiky; provision for 3–5× the daily average, not the average.
Compression is the lever — Parquet + Snappy/Zstd turns 200 GB raw into ~20 GB. Storage is cheap; scan cost dominates the bill.
Compute > storage — a full daily re-transform scanning 7.3 TB at Snowflake/BigQuery rates (~$5/TB scanned) is ~$36/day = ~$13k/yr. Incremental (scan today's 20 GB) is ~$0.10/day. This is why incremental models matter.
Airflow Internals: Scheduler, Executor & Backfill
Airflow is a scheduler that turns a DAG (Python-defined dependency graph) into timed task runs. Understanding the loop explains most production surprises.
Scheduler loop — parses DAG files (the DagFileProcessor subprocess), computes which DagRuns are due, and enqueues TaskInstances whose upstream dependencies are met. It writes state to the metadata DB (Postgres); the DB is the single source of truth.
Executor — pulls queued tasks and runs them. LocalExecutor (subprocesses, single node), CeleryExecutor (worker pool + Redis/RabbitMQ broker), KubernetesExecutor (one pod per task, isolated deps, autoscaling). The scheduler decides what; the executor decides where.
data_interval — a run for interval [start, end) fires after the interval closes. A daily DAG with data_interval_start = 2026-07-01 executes at ~2026-07-02 00:00. Confusing this is the #1 Airflow bug.
Backfill replays historical intervals: airflow dags backfill -s 2026-01-01 -e 2026-06-30 my_dag materializes one DagRun per missed interval. It only works correctly if tasks are idempotent — rerunning interval X must produce the same result, not append duplicates.
# Idempotent by design: partition-overwrite keyed to the interval, not INSERT
@task
def load_partition(**ctx):
ds = ctx["data_interval_start"].format("YYYY-MM-DD")
spark.sql(f"""
INSERT OVERWRITE TABLE silver.events
PARTITION (event_date = '{ds}')
SELECT * FROM bronze.events WHERE event_date = '{ds}'
""")
# Rerun -> same partition overwritten. A naive INSERT would double-count.
Sensors, Deferrable Operators & Dynamic Mapping
Two features separate a toy DAG from a production one: waiting on external state without burning a worker slot, and generating tasks from runtime data.
Mechanism
How it waits
Cost
Sensor (poke)
Occupies a worker slot, polls every N s
1 slot held for hours — deadlock risk
Sensor (reschedule)
Releases slot between pokes, re-queues
Slot free, but scheduler churn each poke
Deferrable operator
Suspends to the triggerer (async event loop), zero slot
~thousands of waits on one triggerer process
# Deferrable: waits on an S3 key using async I/O, holds no worker slot
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
wait = S3KeySensor(
task_id="wait_for_drop",
bucket_key="s3://raw/events/{{ ds }}/_SUCCESS",
deferrable=True, # hands off to the triggerer
poke_interval=60, timeout=6 * 3600,
)
# Dynamic task mapping: fan out one task per file discovered at runtime
@task
def list_files(**ctx) -> list[str]:
return s3.list_keys(bucket="raw", prefix=ctx["ds"])
@task
def process(key: str):
load(key)
process.expand(key=list_files()) # N parallel task instances, N known only at run time
Prefer deferrable for any wait longer than a few minutes — sensors in poke mode silently exhaust the pool and stall the whole cluster.
Dynamic mapping (.expand(), Airflow 2.3+) replaces brittle PythonOperator-generates-subdag hacks; the UI shows a mapped-index grid.
Cap the fan-out — set max_active_tis_per_dag / pool limits so a 10,000-file day doesn't stampede the executor.