The conceptual core of Data Engineering. Once you understand these patterns, Airflow, dbt, and PySpark are just tools for implementing them at scale.
The same transform, both ways — say you need each order's revenue net of a 5% platform fee:
# ETL: computed in Python, BEFORE the warehouse ever sees it
def transform(df):
df["net_revenue"] = df["total_amount"] * 0.95
return df
# by the time this lands in Postgres, only the already-transformed column exists —
# if the fee logic was wrong, you must re-run the whole extract to fix it-- ELT: raw total_amount is loaded as-is, the fee logic lives in a dbt model instead
-- models/marts/fct_orders.sql
SELECT
order_id,
total_amount,
total_amount * 0.95 AS net_revenue
FROM {{ ref('stg_orders') }}
-- raw total_amount is still sitting in the warehouse — if the 0.95 was wrong,
-- fix the SQL and re-run dbt, no need to re-extract anything from the sourceThat's the real argument for ELT, beyond "cloud warehouses are fast now": keeping the raw, untransformed data around means fixing a transformation bug is cheap — just edit the SQL and rerun it. With ETL, fixing the same bug can mean re-extracting from the original source, which might not even have that old data anymore.
Processing data in scheduled chunks (hourly/daily) rather than continuously.
| Batch | Micro-batch | Streaming | |
|---|---|---|---|
| Latency | Hours to a day | Minutes | Seconds or less |
| Typical trigger | Cron schedule (Airflow) | Frequent schedule | Event arrival (Kafka) |
| Complexity | Low | Medium | High |
| Good fit | Daily reporting, warehouse loads | Near-real-time dashboards | Fraud detection, live alerts |
The real question isn't "batch or streaming" in the abstract — it's "what does it actually cost if this data is a few hours old?" A finance report that runs once a day is perfectly fine as batch. A fraud check that only fires after the fraudulent charge has already gone through is useless. Default to batch (Airflow, Week 7-8) unless you can point to a real cost of the delay — streaming (Week 11) adds real operational work that isn't worth taking on just because it sounds more advanced.
-- naive: reload everything every run (simple but wasteful)
SELECT * FROM source_orders;
-- incremental: only pull what changed since the last run
SELECT * FROM source_orders
WHERE updated_at > :last_watermark;Worked example — a small table that tracks the watermark per source, updated only after a successful load:
CREATE TABLE pipeline_watermarks (
source_name TEXT PRIMARY KEY,
last_watermark TIMESTAMPTZ NOT NULL
);
-- 1. read the current watermark
SELECT last_watermark FROM pipeline_watermarks WHERE source_name = 'orders_api';
-- e.g. returns 2026-08-17 02:00:00+00
-- 2. pull only rows changed since then
SELECT * FROM source_orders WHERE updated_at > '2026-08-17 02:00:00+00';
-- 3. only after the load succeeds, advance the watermark to the max updated_at just pulled
UPDATE pipeline_watermarks
SET last_watermark = '2026-08-18 02:00:00+00'
WHERE source_name = 'orders_api';The important detail is that step 3 only runs after the load succeeds. If the pipeline crashes partway through, the watermark never moves forward, so next time it naturally re-pulls the same window instead of quietly skipping data. One common mistake: using NOW() (the current time) instead of the actual latest updated_at you pulled. If a row gets written a few seconds late, a NOW()-based watermark can skip it forever. Always move the watermark forward based on the data you actually saw, not the clock.
Capturing row-level insert/update/delete events from a source database as they happen, instead of polling.
What a CDC event actually looks like — tools like Debezium emit one JSON message per row change, typically onto a Kafka topic (Week 11), carrying the before/after state and an operation code:
{
"op": "u", // c = create, u = update, d = delete, r = initial snapshot read
"before": {"customer_id": 101, "city": "Delhi"},
"after": {"customer_id": 101, "city": "Chicago"},
"source": {"table": "customers", "ts_ms": 1755500000000}
}Having both before and after is what makes CDC so useful for data modeling. That one event already has everything you need to build an SCD Type 2 record (see below) — "Delhi" becomes the old, closed-out row, "Chicago" becomes the new current one — all without ever having to check the source table yourself to notice something changed.
def validate(df):
assert df["order_id"].is_unique, "duplicate order_id found"
assert df["total_amount"].ge(0).all(), "negative order amount found"
assert df["customer_id"].notna().all(), "null customer_id found"
return dftests (Week 7-8) and tools like Great Expectations formalize.Should you stop the whole batch, or just set aside the bad rows? The plain assert example above stops everything the moment one row is bad. That's the right call for serious problems — a duplicate order_id usually means something upstream is badly broken. But it's overkill when only a few rows have small issues and the rest are perfectly fine to load:
def validate_and_quarantine(df):
is_negative = df["total_amount"].lt(0)
is_missing_customer = df["customer_id"].isna()
bad_rows = df[is_negative | is_missing_customer].copy()
good_rows = df[~(is_negative | is_missing_customer)]
if len(bad_rows) > 0:
bad_rows["rejected_reason"] = bad_rows.apply(
lambda r: "negative_amount" if r["total_amount"] < 0 else "missing_customer_id", axis=1
)
bad_rows.to_sql("rejected_orders", engine, if_exists="append", index=False)
logger.warning("Quarantined %d bad rows out of %d", len(bad_rows), len(df))
return good_rowsThis "quarantine" pattern — send bad rows to a separate table with a reason attached, instead of crashing or just silently dropping them — lets the pipeline keep running while still keeping every rejected row visible so you can look at it later. Use a hard failure when something is fundamentally broken, and quarantine for the smaller data-quality issues you expect to see now and then.
-- keep only the latest record per order_id
SELECT * FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) AS rn
FROM raw_orders
) t
WHERE rn = 1;ROW_NUMBER() ... WHERE rn = 1 pattern (straight from your Week 1-2 window functions) is the standard SQL dedup technique.df.drop_duplicates(subset=["order_id"], keep="last").Sometimes "duplicate" needs more than one column to detect. There isn't always a single ID to key on — two rows might only count as duplicates when several fields match together. A classic example: the same customer accidentally placing what looks like the same order twice in a row, from double-clicking "buy" or a retried request:
SELECT * FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY customer_id, total_amount, DATE_TRUNC('minute', order_date)
ORDER BY order_id
) AS rn
FROM raw_orders
) t
WHERE rn = 1;Grouping by several columns together (customer_id + total_amount + the order time rounded to the nearest minute) catches near-duplicates that a simple order_id-only dedupe would completely miss — a retried request usually gets a brand-new order_id, even though it's really the same order.
Running the same pipeline twice on the same input produces the same result — no duplicated or corrupted data.
DELETE the target partition/date range then INSERT (rather than blind INSERT), or use UPSERT (INSERT ... ON CONFLICT DO UPDATE) keyed on a natural/unique key.INSERT INTO orders VALUES (...) unconditionally — re-running it doubles your data.INSERT INTO orders_clean (order_id, total_amount)
VALUES (1, 99.5)
ON CONFLICT (order_id) DO UPDATE SET total_amount = EXCLUDED.total_amount;Seeing the bug happen makes this concrete. A naive daily load script:
-- run once: fine
INSERT INTO orders_clean SELECT * FROM staging_orders WHERE load_date = CURRENT_DATE;
-- Airflow retries this task after a transient network blip during the COMMIT...
-- run again: every row from staging_orders for today is now inserted a SECOND time
-- your revenue numbers are silently ~2x what they should be, with no error raised anywhereThe two idempotent fixes from the bullet above, side by side:
-- fix 1: delete the target window first, so a retry replaces rather than appends
DELETE FROM orders_clean WHERE load_date = CURRENT_DATE;
INSERT INTO orders_clean SELECT * FROM staging_orders WHERE load_date = CURRENT_DATE;
-- fix 2: upsert keyed on a natural/unique key, so a retry updates in place instead of duplicating
INSERT INTO orders_clean (order_id, total_amount, load_date)
SELECT order_id, total_amount, load_date FROM staging_orders WHERE load_date = CURRENT_DATE
ON CONFLICT (order_id) DO UPDATE SET total_amount = EXCLUDED.total_amount;Either version can run once, or 100 times, on the same input and always end up in the exact same final state. That's really what "idempotent" means — and it's what makes Airflow's automatic retries (Week 7-8) safe instead of risky.
| Fact table | Dimension table |
|---|---|
| Measurable events/transactions (orders, payments, clicks) | Descriptive context (customers, products, dates) |
| Numeric, additive columns (amount, quantity) | Mostly text/categorical attributes |
| Grows fast, one row per event | Grows slowly, one row per entity |
| References dimensions via foreign keys | Referenced by facts |
Example: fact_orders (order_id, customer_key, product_key, date_key, amount) joins out to dim_customers, dim_products, dim_date.
Grain — the most important design decision in a fact table. "Grain" means what one row represents. Get it wrong and every downstream metric is wrong:
-- grain = one row per ORDER (order_id is unique)
fact_orders(order_id, customer_key, date_key, order_total)
-- grain = one row per ORDER ITEM (order_id repeats once per line item)
fact_order_items(order_item_id, order_id, product_key, quantity, unit_price)A report asking for "revenue per product" has to use fact_order_items, since that's the one with product-level detail. Trying to answer the same question from fact_orders just isn't possible — that information doesn't exist at the order level. Always write down a fact table's grain (often just as a comment or a dbt doc note) before you add a single column to it.
Numeric columns (called "measures") also fall into three types worth knowing: additive ones are safe to SUM across anything, like order_total. Semi-additive ones are safe to sum in some ways but not others — you can add up an account balance across many accounts, but adding it up across time gives you a meaningless number. Non-additive ones should never be summed at all, like a unit_price or a percentage. Knowing which type a column is stops you from building a report that quietly makes no sense.
Why it's worth the denormalization — compare the query to get "revenue by customer city" against a normalized OLTP schema (Week 1-2's customers/orders tables) versus a star schema:
-- normalized OLTP: fine here, but this pattern compounds fast as more attributes get involved
SELECT c.city, SUM(o.total_amount)
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
GROUP BY c.city;
-- star schema: identical shape, but dim_customers already carries every descriptive
-- attribute analysts need (city, signup cohort, segment...) without touching more tables
SELECT dc.city, SUM(fo.amount)
FROM fact_orders fo
JOIN dim_customers dc ON fo.customer_key = dc.customer_key
GROUP BY dc.city;The real payoff shows up once a report needs five descriptive columns instead of one. In a normalized OLTP schema, those five columns might be spread across three or four different tables, each needing its own join. In a star schema, dim_customers already has all of them sitting on one row per customer — so the query stays a single simple join, no matter how many descriptive columns it ends up using.
Slowly Changing Dimensions — how to handle a dimension attribute (e.g. a customer's address) changing over time.
SCD Type 1 — overwrite: no history kept.
| customer_id | city |
|---|---|
| 101 | Chicago (was Delhi, now overwritten) |
SCD Type 2 — track history with new rows + effective dates + a current flag:
| customer_key | customer_id | city | valid_from | valid_to | is_current |
|---|---|---|---|---|---|
| 1 | 101 | Delhi | 2024-01-01 | 2026-03-14 | false |
| 2 | 101 | Chicago | 2026-03-15 | NULL | true |
customer_key (surrogate key, one per version) vs customer_id (natural/business key, stays the same across versions) — this distinction is exactly what trips people up.snapshot feature (Week 7-8) implements SCD Type 2 automatically.Implementing SCD Type 2 by hand in SQL — worth doing once so dbt's snapshot feature (Week 7-8) doesn't feel like magic later. Given a new value arrives for customer 101's city:
-- 1. close out the current row for this customer (if the value actually changed)
UPDATE dim_customers
SET valid_to = NOW(), is_current = false
WHERE customer_id = 101 AND is_current = true AND city != 'Chicago';
-- 2. insert the new current row, only if step 1 actually closed something out
INSERT INTO dim_customers (customer_id, city, valid_from, valid_to, is_current)
SELECT 101, 'Chicago', NOW(), NULL, true
WHERE EXISTS (
SELECT 1 FROM dim_customers
WHERE customer_id = 101 AND city != 'Chicago' AND valid_to = NOW()
);The WHERE city != 'Chicago' check on both statements is what makes this safe to re-run: running the exact same update a second time does nothing, because there's no longer a row where the city is different. This close-then-insert pattern is exactly what dbt's snapshot does for you automatically on every run — it just uses its own column names, dbt_valid_from/dbt_valid_to, instead of the ones written by hand here.
Extend your Week 3 pipeline: API → Raw Data → Transform → PostgreSQL → Analytics tables
Wiring the layers together — one query per hop, showing exactly what changes shape at each stage:
-- raw_orders: exactly what the API returned, untyped, unvalidated (ELT landing zone)
CREATE TABLE raw_orders (payload JSONB, loaded_at TIMESTAMPTZ DEFAULT NOW());
-- staging_orders: typed, deduped, validated — one clean row per real order
INSERT INTO staging_orders (order_id, customer_id, order_date, total_amount)
SELECT DISTINCT ON ((payload->>'order_id')::int)
(payload->>'order_id')::int,
(payload->>'customer_id')::int,
(payload->>'order_date')::timestamptz,
(payload->>'total_amount')::numeric
FROM raw_orders
WHERE (payload->>'total_amount')::numeric >= 0 -- validation
ORDER BY (payload->>'order_id')::int, loaded_at DESC; -- dedup: keep the latest load per order_id
-- fact_orders: the star-schema shape, keyed to dimensions instead of raw IDs
INSERT INTO fact_orders (order_id, customer_key, date_key, amount)
SELECT s.order_id, dc.customer_key, dd.date_key, s.total_amount
FROM staging_orders s
JOIN dim_customers dc ON dc.customer_id = s.customer_id AND dc.is_current = true
JOIN dim_date dd ON dd.date = s.order_date::date;
-- analytics.customer_ltv: the final reporting aggregate, cheap to query repeatedly
CREATE VIEW analytics.customer_ltv AS
SELECT dc.customer_id, dc.city, SUM(fo.amount) AS lifetime_value
FROM fact_orders fo
JOIN dim_customers dc ON fo.customer_key = dc.customer_key
GROUP BY dc.customer_id, dc.city;DISTINCT ON (a Postgres-only feature) is doing two jobs at once here: it's a shorter way to write the dedup step from earlier, and it handles type-casting in the same pass. Notice the dim_customers join uses is_current = true — that's the SCD Type 2 dimension from above, which means every fact row gets joined to whatever the customer's details were at the time the order happened, not to today's values.
dim_customers) as SCD Type 2, add an idempotent load (upsert on a natural key), and write a validation step that rejects bad rows before they reach staging_orders.