Lab 1 — Build a star schema from a normalized source

Objective

By the end of this lab you will have a local SQLite database, roaster.db, holding two models of the same data: a normalized (3NF) coffee-shop source, and a star schema you design and load from it — one fact table at line-item grain surrounded by three dimensions. You will run an analytical query that reports revenue by category and month, then prove the fact grain holds so no aggregation double-counts. Everything uses Python's standard-library sqlite3 module: no server, no install, no cost.

Skills exercised:

• Reading a normalized source schema and designing a star schema target
• Declaring a fact grain and building dimensions and a fact table for it
• Loading dimensions and a fact from a normalized source with joins
• Proving the grain and reconciling the warehouse against the source

Prerequisites and setup

Prerequisites: the three lessons in this course — Normalization vs. Denormalization, Dimensional Modelling — Star and Snowflake, and Slowly Changing Dimensions and Loading Patterns. DE-102 SQL for Data Engineers is recommended for the join and aggregation fluency this lab assumes.

Setup: you need Python 3.10 or newer. The sqlite3 module ships with Python, so there is nothing to install. Verify your version and create a working folder.

macOS / Linux

python3 --version
mkdir de201-star-lab && cd de201-star-lab

Windows (PowerShell)

python --version
mkdir de201-star-lab; cd de201-star-lab

Expected output from the version command is a line like Python 3.12.10. If the command is not found, install Python from python.org before continuing. In the steps below, use python3 on macOS/Linux and python on Windows.

note

Every step writes a small Python file and reads or writes roaster.db in this one folder. If you close your terminal and come back, re-run the cd de201-star-lab command to return to it; roaster.db and your scripts are still there, so you can resume at the step you left off. The steps build on each other, so run them in order the first time through.

Step 1 — Build the normalized source

Goal: create roaster.db with a normalized, five-table OLTP source — customers, categories, products, orders, and order items — and confirm the row counts.

Save this exactly as build_source.py:

"""Step 1: build roaster.db, a small normalized (3NF) OLTP coffee source."""
import sqlite3

con = sqlite3.connect("roaster.db")
con.executescript("""
DROP TABLE IF EXISTS order_items;
DROP TABLE IF EXISTS orders;
DROP TABLE IF EXISTS products;
DROP TABLE IF EXISTS categories;
DROP TABLE IF EXISTS customers;

CREATE TABLE customers (
    customer_id INTEGER PRIMARY KEY,
    full_name   TEXT NOT NULL,
    country     TEXT NOT NULL,
    signup_date TEXT NOT NULL
);

CREATE TABLE categories (
    category_id   INTEGER PRIMARY KEY,
    category_name TEXT NOT NULL
);

CREATE TABLE products (
    product_id   INTEGER PRIMARY KEY,
    product_name TEXT NOT NULL,
    roast_level  TEXT NOT NULL,
    category_id  INTEGER NOT NULL,
    FOREIGN KEY (category_id) REFERENCES categories(category_id)
);

CREATE TABLE orders (
    order_id    INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    order_date  TEXT NOT NULL,
    status      TEXT NOT NULL,
    FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
);

CREATE TABLE order_items (
    item_id    INTEGER PRIMARY KEY,
    order_id   INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    quantity   INTEGER NOT NULL,
    unit_price REAL NOT NULL,
    FOREIGN KEY (order_id)   REFERENCES orders(order_id),
    FOREIGN KEY (product_id) REFERENCES products(product_id)
);

INSERT INTO customers VALUES
    (1, 'Priya Nair',    'IN', '2025-12-01'),
    (2, 'Tomas Alvarez', 'MX', '2026-01-11'),
    (3, 'Amina Yusuf',   'KE', '2026-02-02'),
    (4, 'Wei Zhang',     'CN', '2026-03-19');

INSERT INTO categories VALUES
    (1, 'Single Origin'),
    (2, 'Blend'),
    (3, 'Equipment');

INSERT INTO products VALUES
    (10, 'Yirgacheffe 250g',   'light',  1),
    (11, 'Sidamo 250g',        'medium', 1),
    (12, 'House Blend 500g',   'dark',   2),
    (13, 'Espresso Blend 1kg', 'dark',   2),
    (14, 'Ceramic Dripper',    'none',   3);

INSERT INTO orders VALUES
    (1001, 1, '2026-06-02', 'completed'),
    (1002, 1, '2026-06-20', 'completed'),
    (1003, 2, '2026-06-05', 'completed'),
    (1004, 3, '2026-07-01', 'completed'),
    (1005, 4, '2026-07-08', 'completed');

INSERT INTO order_items VALUES
    (1, 1001, 10, 2, 14.00),
    (2, 1001, 14, 1, 22.00),
    (3, 1002, 12, 3, 18.00),
    (4, 1003, 11, 1, 15.00),
    (5, 1003, 13, 2, 30.00),
    (6, 1004, 10, 4, 14.00),
    (7, 1005, 12, 1, 18.00),
    (8, 1005, 13, 1, 30.00);
""")
con.commit()

counts = {
    t: con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
    for t in ("customers", "categories", "products", "orders", "order_items")
}
print(counts)
con.close()

Note the shape of the source: the category name lives once in categories and is referenced by products, and the price paid lives on each order_items row. This is the normalized source your star schema will read from.

Checkpoint: run the builder.

python3 build_source.py

You should see exactly:

{'customers': 4, 'categories': 3, 'products': 5, 'orders': 5, 'order_items': 8}

If you see an error about a locked database, close any other program holding roaster.db and re-run. If counts differ, re-copy the INSERT statements — every value row must be present.

Step 2 — Design and create the star schema

Goal: create the empty star schema — three dimensions and one fact table — with the fact grain declared up front as one row per order line item.

The design decisions to notice as you type this:

• dim_product denormalizes the category name in as a plain column, so a category slice needs no extra join — the star pattern from the modelling lesson.
• Each dimension has a surrogate key (customer_key, product_key, date_key) that the fact stores, separate from the source's natural id.
• fact_sales.item_id is declared UNIQUE. That constraint is the grain, enforced: one order line item can land in the fact only once.

Save this as build_star.py:

"""Step 2: create the star schema — three dimensions and one fact table."""
import sqlite3

con = sqlite3.connect("roaster.db")
con.executescript("""
DROP TABLE IF EXISTS fact_sales;
DROP TABLE IF EXISTS dim_customer;
DROP TABLE IF EXISTS dim_product;
DROP TABLE IF EXISTS dim_date;

CREATE TABLE dim_customer (
    customer_key INTEGER PRIMARY KEY AUTOINCREMENT,
    customer_id  INTEGER NOT NULL UNIQUE,
    full_name    TEXT NOT NULL,
    country      TEXT NOT NULL
);

CREATE TABLE dim_product (
    product_key   INTEGER PRIMARY KEY AUTOINCREMENT,
    product_id    INTEGER NOT NULL UNIQUE,
    product_name  TEXT NOT NULL,
    roast_level   TEXT NOT NULL,
    category_name TEXT NOT NULL
);

CREATE TABLE dim_date (
    date_key   INTEGER PRIMARY KEY,
    full_date  TEXT NOT NULL UNIQUE,
    year       INTEGER NOT NULL,
    month      INTEGER NOT NULL,
    month_name TEXT NOT NULL
);

-- Grain: one row per order line item (one product on one order).
CREATE TABLE fact_sales (
    sale_key     INTEGER PRIMARY KEY AUTOINCREMENT,
    item_id      INTEGER NOT NULL UNIQUE,
    order_id     INTEGER NOT NULL,
    customer_key INTEGER NOT NULL,
    product_key  INTEGER NOT NULL,
    date_key     INTEGER NOT NULL,
    quantity     INTEGER NOT NULL,
    unit_price   REAL NOT NULL,
    line_revenue REAL NOT NULL,
    FOREIGN KEY (customer_key) REFERENCES dim_customer(customer_key),
    FOREIGN KEY (product_key)  REFERENCES dim_product(product_key),
    FOREIGN KEY (date_key)     REFERENCES dim_date(date_key)
);
""")
con.commit()

star = [
    r[0] for r in con.execute("""
        SELECT name FROM sqlite_master
        WHERE type = 'table' AND name LIKE 'dim_%' OR name = 'fact_sales'
        ORDER BY name
    """).fetchall()
]
print(star)
con.close()

Checkpoint: run it.

python3 build_star.py

You should see exactly the four star tables, and no error:

['dim_customer', 'dim_date', 'dim_product', 'fact_sales']

The tables are empty for now — you load them in the next two steps. If a table is missing from the list, re-check that its CREATE TABLE statement ran without a syntax error.

Step 3 — Load the dimensions

Goal: fill the three dimensions from the normalized source, denormalizing the category name into dim_product and deriving dim_date from the order dates.

Load dimensions before the fact — the fact needs their surrogate keys to point at. Save this as load_dimensions.py:

"""Step 3: load the three dimensions from the normalized source."""
import sqlite3

con = sqlite3.connect("roaster.db")

con.execute("""
    INSERT INTO dim_customer (customer_id, full_name, country)
    SELECT customer_id, full_name, country
    FROM customers
""")

# dim_product denormalizes the category name in — a flat star dimension.
con.execute("""
    INSERT INTO dim_product (product_id, product_name, roast_level, category_name)
    SELECT p.product_id, p.product_name, p.roast_level, c.category_name
    FROM products   p
    JOIN categories c ON c.category_id = p.category_id
""")

# dim_date: one row per distinct order date, with parts split out for slicing.
con.execute("""
    INSERT INTO dim_date (date_key, full_date, year, month, month_name)
    SELECT DISTINCT
        CAST(strftime('%Y%m%d', order_date) AS INTEGER),
        order_date,
        CAST(strftime('%Y', order_date) AS INTEGER),
        CAST(strftime('%m', order_date) AS INTEGER),
        CASE strftime('%m', order_date)
            WHEN '06' THEN 'June'
            WHEN '07' THEN 'July'
        END
    FROM orders
""")
con.commit()

counts = {
    t: con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
    for t in ("dim_customer", "dim_product", "dim_date")
}
print(counts)
con.close()

Checkpoint: run it.

python3 load_dimensions.py

You should see exactly:

{'dim_customer': 4, 'dim_product': 5, 'dim_date': 5}

Four customers, five products, and five distinct order dates. If dim_date shows fewer than 5, two orders shared a date — check the orders rows from Step 1. If you get a UNIQUE constraint failed error, you ran this script twice; re-run build_star.py to reset the empty dimensions, then run this once.

Step 4 — Load the fact at line-item grain

Goal: load fact_sales with one row per completed order line item, resolving each source id to its dimension surrogate key and computing line_revenue.

Save this as load_fact.py:

"""Step 4: load fact_sales at one row per order line item."""
import sqlite3

con = sqlite3.connect("roaster.db")

con.execute("""
    INSERT INTO fact_sales (
        item_id, order_id, customer_key, product_key, date_key,
        quantity, unit_price, line_revenue
    )
    SELECT
        oi.item_id,
        oi.order_id,
        dc.customer_key,
        dp.product_key,
        dd.date_key,
        oi.quantity,
        oi.unit_price,
        ROUND(oi.quantity * oi.unit_price, 2)
    FROM order_items  oi
    JOIN orders       o  ON o.order_id     = oi.order_id
    JOIN dim_customer dc ON dc.customer_id = o.customer_id
    JOIN dim_product  dp ON dp.product_id  = oi.product_id
    JOIN dim_date     dd ON dd.full_date   = o.order_date
    WHERE o.status = 'completed'
""")
con.commit()

fact_rows = con.execute("SELECT COUNT(*) FROM fact_sales").fetchone()[0]
total_revenue = con.execute(
    "SELECT ROUND(SUM(line_revenue), 2) FROM fact_sales"
).fetchone()[0]
print("fact rows:", fact_rows)
print("total revenue:", total_revenue)
con.close()

The load joins each order item back through orders to reach dim_customer, and directly to dim_product and dim_date, swapping every natural id for its surrogate key. line_revenue is the additive measure quantity * unit_price, computed once at load time.

Checkpoint: run it.

python3 load_fact.py

You should see exactly:

fact rows: 8
total revenue: 283.0

Eight fact rows — one per source line item — and total revenue of 283.0. If you get a UNIQUE constraint failed: fact_sales.item_id error, the fact already holds rows from a previous run; re-run build_star.py and load_dimensions.py to reset, then run this once. If fact rows is fewer than 8, a join dropped rows — check that Step 3 loaded all dimensions.

Step 5 — Run the analytical query and prove the grain

Goal: answer a real question — revenue by category and month — and prove the fact grain holds, so you can trust the SUM.

Save this as analyze.py:

"""Step 5: analytical query, plus a proof that the fact grain holds."""
import sqlite3

con = sqlite3.connect("roaster.db")

print("Revenue by category and month:")
rows = con.execute("""
    SELECT dp.category_name,
           dd.month_name,
           ROUND(SUM(fs.line_revenue), 2) AS revenue,
           SUM(fs.quantity)               AS units
    FROM fact_sales  fs
    JOIN dim_product dp ON dp.product_key = fs.product_key
    JOIN dim_date    dd ON dd.date_key    = fs.date_key
    GROUP BY dp.category_name, dd.month, dd.month_name
    ORDER BY dd.month, revenue DESC
""").fetchall()
for row in rows:
    print(" ", row)

# Grain proof: no item_id appears more than once in the fact table.
max_per_item = con.execute("""
    SELECT MAX(c) FROM (
        SELECT COUNT(*) AS c FROM fact_sales GROUP BY item_id
    )
""").fetchone()[0]
print("max fact rows per item_id:", max_per_item)
con.close()

Checkpoint: run it.

python3 analyze.py

You should see exactly:

Revenue by category and month:
  ('Blend', 'June', 114.0, 5)
  ('Single Origin', 'June', 43.0, 3)
  ('Equipment', 'June', 22.0, 1)
  ('Single Origin', 'July', 56.0, 4)
  ('Blend', 'July', 48.0, 2)
max fact rows per item_id: 1

The report slices revenue by a category_name that lives only in dim_product and a month_name that lives only in dim_date — two dimension joins, no category or date logic in the fact. The final line is the grain proof: a maximum of 1 fact row per item_id means the declared grain holds, so the SUM cannot double-count. If that number is greater than 1, the fact was loaded twice — reset with build_star.py, load_dimensions.py, and load_fact.py.

Validation

Verify the whole build against the objective: the grain is one row per line item, the fact row count and revenue reconcile with the source, and every fact key resolves to a dimension. Save this as validate.py:

"""Validation: verify the star schema against the objective, end to end."""
import sqlite3

con = sqlite3.connect("roaster.db")

# 1) Grain: every completed source line item is exactly one fact row.
fact_rows = con.execute("SELECT COUNT(*) FROM fact_sales").fetchone()[0]
source_lines = con.execute("""
    SELECT COUNT(*) FROM order_items oi
    JOIN orders o ON o.order_id = oi.order_id
    WHERE o.status = 'completed'
""").fetchone()[0]
max_per_item = con.execute("""
    SELECT MAX(c) FROM (
        SELECT COUNT(*) AS c FROM fact_sales GROUP BY item_id
    )
""").fetchone()[0]

# 2) Reconciliation: fact revenue equals the source's completed revenue.
fact_revenue = con.execute(
    "SELECT ROUND(SUM(line_revenue), 2) FROM fact_sales"
).fetchone()[0]
source_revenue = con.execute("""
    SELECT ROUND(SUM(oi.quantity * oi.unit_price), 2)
    FROM order_items oi
    JOIN orders o ON o.order_id = oi.order_id
    WHERE o.status = 'completed'
""").fetchone()[0]

# 3) Every fact key resolves to a dimension row (no orphans).
orphans = con.execute("""
    SELECT COUNT(*) FROM fact_sales fs
    LEFT JOIN dim_customer dc ON dc.customer_key = fs.customer_key
    LEFT JOIN dim_product  dp ON dp.product_key  = fs.product_key
    LEFT JOIN dim_date     dd ON dd.date_key     = fs.date_key
    WHERE dc.customer_key IS NULL
       OR dp.product_key  IS NULL
       OR dd.date_key     IS NULL
""").fetchone()[0]

con.close()

checks = [
    ("fact grain is one row per line item", max_per_item == 1),
    ("fact row count matches source line items", fact_rows == source_lines),
    ("fact revenue reconciles with source", fact_revenue == source_revenue),
    ("every fact key resolves to a dimension", orphans == 0),
]
for label, ok in checks:
    print(f"  [{'x' if ok else ' '}] {label}")

if all(ok for _, ok in checks):
    print("PASSED - star schema built and grain proven")
else:
    print("FAILED - review the checks above")

Checkpoint: run it.

python3 validate.py

Every check should be marked, and the last line should read:

  [x] fact grain is one row per line item
  [x] fact row count matches source line items
  [x] fact revenue reconciles with source
  [x] every fact key resolves to a dimension
PASSED - star schema built and grain proven

If any line prints FAILED, the unmarked check names the problem: a grain or row-count failure points back to Step 4's fact load (reset and load once), a reconciliation failure means the line_revenue computation drifted from the source, and an orphan means a dimension was not loaded before the fact in Step 3.

Teardown

Nothing is billable — this lab runs entirely on your machine. Keep the de201-star-lab folder if you want to reuse roaster.db and the scripts as a reference for the warehouse-design and layering modules; otherwise delete it with rm -rf de201-star-lab (macOS/Linux) or Remove-Item -Recurse de201-star-lab (Windows PowerShell).

Troubleshooting

Symptom	Likely cause	Fix
sqlite3.OperationalError: no such table: fact_sales	Running a load before Step 2 created the star	Run build_star.py first, from the de201-star-lab folder
sqlite3.IntegrityError: UNIQUE constraint failed: fact_sales.item_id	Ran load_fact.py twice into a filled fact	Re-run build_star.py, then load_dimensions.py, then load_fact.py once each
dim_date count is fewer than 5 in Step 3	Two orders share an order date	Re-check the orders INSERT rows in Step 1; five distinct dates are expected
Step 4 prints fewer than 8 fact rows	A dimension join dropped rows	Confirm Step 3 loaded all three dimensions before running the fact load
Validation reports revenue does not reconcile	line_revenue not quantity * unit_price	Re-check the ROUND(oi.quantity * oi.unit_price, 2) expression in Step 4
strftime returns None for month_name	An order month outside June/July was added	Extend the CASE in Step 3 to map the new month number to its name