Project: Build a Clean Analytical Dataset and Star Schema

Checkpoint acceptance criteria

Use synthetic data or a dataset with clear reuse rights. Preserve raw input, make transformations explicit and prove that a rerun produces the same target state.

• Validate required fields and quarantine rejected rows with reasons.
• Impute or transform only under documented, fitted rules.
• Create dimension and fact tables with integrity constraints.
• Reconcile source, accepted, rejected and loaded counts.

An ETL run should be explainable after it fails

Give every run an identifier and timestamp. Record input filename, checksum, row count and code version. These details connect a warehouse value back to the file and transformation that produced it.

Validate before coercion. If a negative quantity becomes missing during numeric conversion, the reject reason should still say ‘negative quantity’ or ‘invalid quantity text’. Generic missingness after transformation loses the original defect.

Keep rejected rows outside the fact table but inside the audit trail, subject to the same security and retention rules as source data. Never expose sensitive raw values in public logs. Store a row key and reason when full rejected records are unnecessary.

Load dimensions before facts, but do so within a controlled transaction or staged swap. Unique business keys and fact identifiers make reruns safe. An upsert policy must define whether corrections replace, version or reverse earlier facts.

Reconciliation is arithmetic plus meaning. Source rows should equal accepted plus rejected rows. Loaded facts may differ from accepted input only under a documented aggregation grain. Compare sums for additive measures as well as counts.

Validate records and produce two outputs

This Python core demonstrates accepted and rejected flows. A complete project loads the accepted output into the SQL design from the prior lesson.

Install the lesson dependencies

python -m pip install pandas

Write clean rows and rejection reasons

import pandas as pd

raw = pd.DataFrame([
    {"order_line_id": 1, "customer_id": "C1", "quantity": "2", "net_amount": "19.90"},
    {"order_line_id": 2, "customer_id": "",   "quantity": "1", "net_amount": "8.00"},
    {"order_line_id": 3, "customer_id": "C2", "quantity": "-4", "net_amount": "12.00"},
    {"order_line_id": 4, "customer_id": "C3", "quantity": "3", "net_amount": "bad"},
])

working = raw.copy()
working["quantity_num"] = pd.to_numeric(working["quantity"], errors="coerce")
working["amount_num"] = pd.to_numeric(working["net_amount"], errors="coerce")

def reason(row):
    problems = []
    if not str(row["customer_id"]).strip():
        problems.append("missing customer")
    if pd.isna(row["quantity_num"]) or row["quantity_num"] <= 0:
        problems.append("invalid quantity")
    if pd.isna(row["amount_num"]):
        problems.append("invalid amount")
    return "; ".join(problems)

working["reject_reason"] = working.apply(reason, axis=1)
accepted = working[working["reject_reason"] == ""].copy()
rejected = working[working["reject_reason"] != ""].copy()

assert len(raw) == len(accepted) + len(rejected)
assert accepted["order_line_id"].is_unique

accepted.to_csv("accepted-order-lines.csv", index=False)
rejected[["order_line_id", "reject_reason"]].to_csv("rejected-order-lines.csv", index=False)
print({"source": len(raw), "accepted": len(accepted), "rejected": len(rejected)})
print(rejected[["order_line_id", "reject_reason"]].to_string(index=False))

Expected reconciliation

{'source': 4, 'accepted': 1, 'rejected': 3}
 order_line_id     reject_reason
             2  missing customer
             3   invalid quantity
             4     invalid amount

The rejection output intentionally avoids copying every raw field. A production design should decide what investigators need and protect both accepted and rejected records.

Checkpoint failure drills

Run the pipeline twice, inject duplicates and interrupt a load. Recovery behaviour belongs in the project, not in a future operations document.

• A duplicate business key should be rejected or handled by an explicit correction policy.
• An unknown dimension member should be staged, mapped to a declared unknown key or rejected, never silently lost in an inner join.
• A partial database load should roll back or publish only after all validation passes.
• A changed raw file under the same name should produce a different checksum and a new run record.

Complete the end-to-end checkpoint

Extend the example to at least three dimensions and one fact. Generate enough synthetic records to test missing, duplicate, late and corrected data.

• Create schema DDL, a staged loader and an idempotent publish step.
• Save a reject summary by reason without exposing unnecessary values.
• Write count, key-coverage and amount-sum reconciliation queries.
• Run a month-by-region OLAP query and explain which facts it includes.

What to show in a portfolio

• Pipeline diagram and fact grain.
• Synthetic fixtures plus automated validation tests.
• A run report proving rerun safety and reconciliation.

Official references and further reading

• pandas input and output (Official file IO behaviour)
• PostgreSQL INSERT (Official conflict-handling reference)
• PostgreSQL transactions (Official transaction tutorial)

Review note for Project: Build a Clean Analytical Dataset and Star Schema: recheck the linked documentation after a dependency changes the relevant API, metric or modelling assumption, then record the tested version beside your result.