Module 23: Real-World Projects

Goal: a clean, typed, deduplicated sales table from a raw CSV.

base = use_project("proj_01")
(spark.table(f"{catalog}.dbx_course.orders_raw").limit(500)
      .coalesce(1).write.mode("overwrite").option("header", True).csv(f"{base}/in"))   # your "raw file"

schema = "order_id INT, customer_id INT, product_id INT, quantity INT, status STRING, order_ts TIMESTAMP"
raw = spark.read.option("header", True).schema(schema).csv(f"{base}/in")

clean = (raw.dropDuplicates(["order_id"])
            .withColumn("status", F.lower(F.trim("status")))
            .filter("customer_id IS NOT NULL AND quantity > 0"))
clean.write.mode("overwrite").saveAsTable("sales")

Do:

• Compare raw.count() and clean.count(). Write down why they differ
• Save rejected rows to sales_rejected with a reason (Module 21.8)
• Add COMMENTs to the table and columns

Done when: sales has no duplicate order_id, the schema is typed, and the rejected rows are explained.

Goal: five answers and one dashboard from orders, products and customers.

%sql
USE SCHEMA dbx_course;

-- 1. revenue by category and month
SELECT date_trunc('month', o.order_ts) AS month, p.category,
       round(sum(o.quantity * p.unit_price), 2) AS revenue
FROM orders o JOIN products p USING (product_id)
WHERE o.status IN ('completed', 'shipped')
GROUP BY ALL ORDER BY month, category

Do:

• Write each as a view with a clear name and comment
• Build an AI/BI dashboard (Module 13.9): a line chart, a bar chart, a counter and a filter
• State your revenue definition (which statuses) in the dashboard text

Done when: every number reconciles with a simple count(*) check, and the dashboard filters work.

Goal: a pipeline that handles a new batch of files with one rerun.

base = use_project("proj_03")
landing = f"{base}/landing"

def land(batch, lo, hi):
    (spark.table(f"{catalog}.dbx_course.orders").filter(F.col("order_id").between(lo, hi))
          .coalesce(1).write.mode("overwrite").json(f"{landing}/batch_{batch}"))

land(1, 1, 1000)

(spark.readStream.format("cloudFiles").option("cloudFiles.format", "json")
      .option("cloudFiles.inferColumnTypes", "true")
      .option("cloudFiles.schemaLocation", f"{base}/_schema/bronze")
      .option("pathGlobFilter", "*.json").load(landing)
      .withColumn("_ingested_at", F.current_timestamp())
      .writeStream.option("checkpointLocation", f"{base}/_cp/bronze")
      .trigger(availableNow=True).toTable("bronze_orders")).awaitTermination()

Do:

• Build silver_orders (dedupe, clean, is_valid) and silver_quarantine
• Build gold_daily_revenue with a replaceWhere by date (Module 8.13)
• Land batch 2 (land(2, 1001, 2000)) and rerun: only new rows should flow
• Put the three steps in a job with dependencies (Module 10) and a failure email

Done when: a rerun with no new files changes nothing, and bronze = silver + quarantine after dedupe.

Goal: a re-runnable job that fetches only new records.

Free Edition limits outbound access to trusted domains, so the API is simulated. The loader code is the same for a real requests.get.

base = use_project("proj_04")
import json, time

ALL = [{"id": i, "updated_at": f"2025-03-{1 + i // 40:02d}T00:00:00", "value": i * 3} for i in range(1, 121)]

def fake_api(updated_since, page, size=25):
    rows = [r for r in ALL if r["updated_at"] > updated_since]
    chunk = rows[(page - 1) * size: page * size]
    return {"items": chunk, "has_more": page * size < len(rows)}

def fetch_all(updated_since):
    page, out = 1, []
    while True:
        resp = fake_api(updated_since, page)            # replace with requests.get(...).json()
        out += resp["items"]
        if not resp["has_more"]:
            return out
        page += 1
        time.sleep(0.1)                                 # be polite: respect rate limits

Do:

• Store the watermark in a table (Module 21.2) and move it only after the write succeeds
• Write each pull as a raw JSON file (run_<timestamp>.json), then ingest to bronze: the raw copy lets you replay
• Add retry with backoff around the call (Module 21.9). Store the API key in a secret scope
• MERGE into silver on id so repeats don't duplicate

Done when: running twice in a row loads nothing the second time, and an interrupted run recovers without gaps.

Goal: keep a customers_current table identical to a source, from change events only.

base = use_project("proj_05")

changes = spark.createDataFrame([
    (1, "Ada",   "NL", "INSERT", 1), (2, "Bo",    "DE", "INSERT", 1),
    (1, "Ada",   "FR", "UPDATE", 2), (3, "Cy",    "US", "INSERT", 2),
    (2, None,    None, "DELETE", 3), (1, "Ada L.", "FR", "UPDATE", 4),
    (1, "Ada",   "FR", "UPDATE", 2),                       # duplicate event
], "customer_id INT, name STRING, country STRING, op STRING, seq INT")
changes.write.mode("overwrite").saveAsTable("customer_changes")

Do:

• Create customers_current and apply the feed with MERGE: dedupe per key by the highest seq, then DELETE / UPDATE / INSERT (Module 21.3)
• Expected result: customers 1 (Ada L., FR) and 3 (Cy, US). Customer 2 gone
• Apply a second batch of events including an out-of-order one (lower seq than already applied)
• Rebuild the same table with AUTO CDC in a pipeline (Module 9.9) and compare

Done when: applying the feed twice leaves the table unchanged, and an old event never overwrites a newer state.

Goal: dim_customer_history with valid_from, valid_to, is_current, and a point-in-time join.

Do:

• Build the table and apply changes with the two-step MERGE from Module 8.10
• Write the invariant tests (Module 14.9):

h = spark.table("dim_customer_scd2")        # from Module 8.10, or your own table
assert h.filter("is_current").groupBy("customer_id").count().filter("count > 1").count() == 0, "two current rows"
overlap = (h.alias("a").join(h.alias("b"),
           (F.col("a.customer_id") == F.col("b.customer_id")) &
           (F.col("a.valid_from") < F.col("b.valid_from")) &
           (F.col("b.valid_from") < F.coalesce(F.col("a.valid_to"), F.lit("9999-12-31").cast("date")))))
assert overlap.count() == 0, "overlapping validity periods"

• Join orders to the history at order date: each order must see the country valid then
• Apply the same changes with AUTO CDC ... STORED AS SCD TYPE 2 and compare row for row

Done when: one current row per customer, no overlaps, and a fact reports the old country for old orders.

Goal: a stream that survives restarts and bad records, run as scheduled availableNow batches.

• Kafka isn't reachable from Free Edition: use the file source and generator from Module 11
• In production the bronze read becomes spark.readStream.format("kafka")... with value cast to string and parsed with from_json (Module 11.12)

Do:

• Build bronze, silver (dropDuplicatesWithinWatermark) and gold (5-minute windows) as in Module 11.15
• Add a quarantine stream for records that fail parsing (from_json returns NULL fields, or use _corrupt_record)
• Land a late event and a duplicate and observe what happens (Module 11.11)
• Kill and restart a run (stop the cell mid-way) and verify there are no duplicates or gaps
• Create a job that runs the three streams every 10 minutes

Done when: count of silver rows equals distinct event_id in bronze, and a restart changes nothing.

Goal: one function that loads any source listed in a control table, with validation and logging.

base = use_project("proj_08")

spark.sql("""
  CREATE OR REPLACE TABLE etl_meta AS SELECT * FROM VALUES
    ('customers', 'tgt_customers', 'customer_id', 'country IS NOT NULL', 'overwrite', true),
    ('products',  'tgt_products',  'product_id',  'unit_price > 0',      'overwrite', true)
  AS t(source, target, key, quality_rule, write_mode, enabled)""")

def run_source(m):
    df = spark.table(f"{catalog}.dbx_course.{m.source}")
    n_in = df.count()
    good = df.filter(m.quality_rule).dropDuplicates([m.key])
    n_out = good.count()
    good.write.mode(m.write_mode).saveAsTable(m.target)
    return (m.source, n_in, n_out, n_in - n_out)

results = [run_source(r) for r in spark.table("etl_meta").filter("enabled").collect()]

Do:

• Store results in etl_run_log with a timestamp and status. Catch errors per source so one failure doesn't stop the others
• Support append and merge modes with a key
• Add a new source by inserting one row. Don't touch the code
• Validate the metadata itself (missing column, unknown mode) before running
• Run it as a For each job task, one iteration per source (Module 10.10)

Done when: a new source needs only a metadata row, and failures are logged without stopping the run.

Goal: a reusable runner that evaluates rules on any table and stores the outcome.

base = use_project("proj_09")

spark.createDataFrame([
    ("orders",    "order_id not null",   "order_id IS NOT NULL",                 "fail"),
    ("orders",    "quantity positive",   "quantity IS NULL OR quantity > 0",     "warn"),
    ("customers", "country present",     "country IS NOT NULL",                  "warn"),
    ("products",  "price positive",      "unit_price > 0",                       "fail"),
], "table_name STRING, rule STRING, expression STRING, severity STRING"
).write.mode("overwrite").saveAsTable("dq_rules")

def run_rules():
    out = []
    for r in spark.table("dq_rules").collect():
        df = spark.table(f"{catalog}.dbx_course.{r.table_name}")
        total = df.count()
        failed = df.filter(f"NOT ({r.expression})").count()
        out.append((r.table_name, r.rule, r.severity, total, failed))
    res = spark.createDataFrame(out, "table_name STRING, rule STRING, severity STRING, total LONG, failed LONG") \
               .withColumn("checked_at", F.current_timestamp())
    res.write.mode("append").saveAsTable("dq_results")
    return res

res = run_rules()
blockers = res.filter("severity = 'fail' AND failed > 0").count()
assert blockers == 0, f"{blockers} blocking data quality rules failed"

Do:

• Add failed row samples (store a few offending keys)
• Build a dashboard: pass rate per table over time, and failures by severity
• Add a SQL alert that triggers when a fail rule has failures
• Wire it into a job: run the rules before the downstream tasks, and stop on fail
• Add a freshness rule (max(order_ts) not older than X)

Done when: adding a rule is one row, results are queryable over time, and a failing rule stops the job.

Goal: every pull request is tested, and every merge to main deploys.

Do:

• Create a repo with the layout from Module 15.4: src/, tests/, resources/, databricks.yml
• Write 3 unit tests for a transformation function (Module 14.1)
• Define a job in the bundle with a dev and prod target, parameterised by schema (Module 15.11)
• Add the GitHub Actions workflows from Module 15.8
• Deploy to dev from your machine first: databricks bundle validate, then deploy, then run
• Tag a release, then roll back by deploying the previous tag (Module 15.14)

Done when: a failing test blocks the merge, and a merge deploys without manual steps.

Goal: a trustworthy ledger from a stream of transactions, with sensitive data protected.

base = use_project("proj_11")

tx = (spark.range(0, 20000).select(
        F.concat(F.lit("TX"), F.lpad(F.col("id").cast("string"), 8, "0")).alias("tx_id"),
        (F.pmod(F.hash("id", F.lit(1)), F.lit(500)) + 1).alias("account_id"),
        F.when(F.pmod(F.hash("id", F.lit(2)), F.lit(3)) == 0, "CREDIT").otherwise("DEBIT").alias("direction"),
        F.round(F.pmod(F.hash("id", F.lit(3)), F.lit(50000)) / 100 + 1, 2).alias("amount"),
        F.timestamp_seconds(F.lit(1740787200) + F.col("id") * 30).alias("tx_ts")))
tx.write.mode("overwrite").saveAsTable("tx_bronze")

Do:

• Use DECIMAL(18,2) for amounts in the real design. Never floats for money
• Add duplicate and out-of-order transactions to your generator and prove the ledger stays correct
• Write the runbook and the readiness checklist (Module 22.13)

Done when: the reconciliation query returns zero differences, and a replay produces an identical ledger.

Goal: a design document for a platform serving several domains. Nothing to run: this project is about decisions.

Write each section on one page:

Do:

• Use Modules 17, 18, 19 and 22 as input. Reference docs for every platform claim
• Include one diagram per major section
• Have someone else review it against the well-architected pillars (Module 19.12)

Done when: a reader can answer "who owns this data, where does it run, what happens when it fails, and what does it cost?" from the document alone.