Project: Raw Events to an Analytics Mart

This capstone combines the four Data Engineering batches. Raw JSON Lines files contain duplicate and late deliveries. A local DuckDB pipeline preserves raw deliveries, stages one current record per event, publishes dimensions and facts, builds a daily regional mart, checks quality, handles one additive schema change, and records lineage—without credentials or hosted infrastructure.

Prerequisites

Complete Warehouse Cost and Query Performance. You should also understand Incremental Processing, Slowly Changing Dimensions, and dbt Data Tests.

Learning goals

• Integrate local files, SQL transformations, dimensional models, and quality gates.
• Make a rerun safe through stable delivery and event keys.
• Handle a nullable additive currency field without inventing historical values.
• Preserve a failed quality assertion as evidence while publishing a repaired fact grain.

Worked example

The raw grain is one delivered file row; duplicates are legitimate observations there. The fact grain is one accepted event_id. The code defaults absent currency to NULL, not USD, because old records do not prove a currency. Customer region is a small capstone fixture, not inferred from events.

import json
from decimal import Decimal
from pathlib import Path

import duckdb

root = Path("fixtures/data-engineering/pipelines")
files = ("events-batch-1.jsonl", "events-batch-2.jsonl", "events-late.jsonl")
regions = {"c-1": "North", "c-2": "South", "c-3": "North"}
db = duckdb.connect(":memory:")
db.execute("""
CREATE TABLE raw_events(
  delivery_key VARCHAR PRIMARY KEY,
  event_id VARCHAR NOT NULL,
  occurred_at TIMESTAMP NOT NULL,
  arrived_at TIMESTAMP NOT NULL,
  customer_id VARCHAR NOT NULL,
  amount VARCHAR NOT NULL,
  currency VARCHAR
);
""")

def build():
    for filename in files:
        rows = [json.loads(line) for line in (root / filename).read_text().splitlines()]
        for line_number, event in enumerate(rows, start=1):
            db.execute(
                "INSERT OR REPLACE INTO raw_events VALUES (?, ?, ?, ?, ?, ?, ?)",
                [f"{filename}:{line_number}", event["event_id"], event["occurred_at"],
                 event["arrived_at"], event["customer_id"], event["amount"],
                 event.get("currency")],
            )

    duplicate = db.execute("""
      SELECT event_id FROM raw_events GROUP BY event_id HAVING COUNT(*) > 1
      ORDER BY event_id
    """).fetchone()
    print("quality_failure duplicate event_id", duplicate[0])

    db.execute("""
      CREATE OR REPLACE TABLE stg_events AS
      SELECT * EXCLUDE (delivery_key, keep_rank)
      FROM (
        SELECT *, ROW_NUMBER() OVER (
          PARTITION BY event_id ORDER BY arrived_at DESC, delivery_key DESC
        ) AS keep_rank
        FROM raw_events
      ) WHERE keep_rank = 1
    """)
    db.execute("""
      CREATE OR REPLACE TABLE dim_customer AS
      SELECT ROW_NUMBER() OVER (ORDER BY customer_id) AS customer_key,
             customer_id,
             CASE customer_id WHEN 'c-1' THEN 'North'
                              WHEN 'c-2' THEN 'South'
                              WHEN 'c-3' THEN 'North' END AS region
      FROM (SELECT DISTINCT customer_id FROM stg_events)
    """)
    db.execute("""
      CREATE OR REPLACE TABLE fct_events AS
      SELECT s.event_id, d.customer_key, s.occurred_at, s.arrived_at,
             CAST(ROUND(CAST(s.amount AS DECIMAL(12,2)) * 100) AS INTEGER) AS amount_cents,
             s.currency
      FROM stg_events AS s JOIN dim_customer AS d USING (customer_id)
    """)
    db.execute("""
      CREATE OR REPLACE TABLE mart_daily_region AS
      SELECT CAST(f.occurred_at AS DATE) AS event_date, d.region,
             COUNT(*) AS event_count, SUM(f.amount_cents)::INTEGER AS amount_cents
      FROM fct_events AS f JOIN dim_customer AS d USING (customer_key)
      GROUP BY event_date, d.region
    """)
    rows, total = db.execute(
        "SELECT COUNT(*), SUM(amount_cents)::INTEGER FROM fct_events"
    ).fetchone()
    assert rows == 6 and total == 4550
    assert db.execute("SELECT COUNT(*) = COUNT(DISTINCT event_id) FROM fct_events").fetchone()[0]
    return rows, total

print("first_run", *build())
print("rerun", *build())
customer_rows = db.execute("SELECT customer_id FROM dim_customer").fetchall()
assert set(regions) == {row[0] for row in customer_rows}
print("schema_change currency")
print("lineage raw_events -> stg_events -> fct_events -> mart_daily_region")

The first run and rerun both publish six facts totaling 4,550 cents. The raw uniqueness assertion deliberately fails on e-101; the deterministic latest-arrival rule then restores fact grain. In production, confirm that this rule matches source semantics—latest delivery is not automatically the authoritative correction.

The additive currency column remains null for historical records. Before mixing currencies, quarantine unknowns or convert with governed rates and dates. A breaking change to units, grain, or keys needs a versioned migration or full rebuild, not merely CREATE OR REPLACE.

Limitations and common mistakes

• In-memory publication is atomic only within this demonstration; production needs durable transactions and recovery evidence.
• Hard-coded regions stand in for a governed dimension source.
• Deduplication can hide upstream delivery regressions unless duplicate metrics remain visible.
• The capstone has no deletes, exchange-rate logic, access controls, or personally identifiable information.
• A passing reconciliation proves the fixture contract, not completeness of an external source.

Practice

Add a corrected late event and a real currency value. Predict which raw, staged, fact, and mart rows change; add a quality gate that prevents mixed-currency aggregation.

Next steps

You have completed the 36-page Data Engineering path. Revisit Relational Data Models for the foundations, Choosing a Data Processing Tool for workload decisions, or Pipeline Observability to turn the capstone checks into operational signals.

Primary sources

Checked 2026-10-15: DuckDB Python API, DuckDB INSERT, dbt incremental models, dbt data tests, and manifest lineage metadata.