Build a Star Schema with Python ETL

Transform flat retail data into a Kimball star schema data warehouse.

Introduction

30 Second Summary

Every company stores sales data, but the difference between a messy spreadsheet and a proper data warehouse is what separates junior analysts from data engineers.

In this project, you will build a complete ETL pipeline in Python that transforms flat, denormalized retail sales data into a Kimball star schema data warehouse running on PostgreSQL.

What You'll Build

A local data warehouse you can query with clean, readable SQL to answer business questions like "What's our monthly revenue by category?" and "Who are our top customers?" in just a few lines of code.

By the end of this project, you'll have:

  • A working Python ETL pipeline that extracts 500 sales records from a staging area, transforms them into proper dimensions and facts with surrogate keys, and loads them into a star schema.
  • Clean analytical queries (monthly revenue, top customers, category performance, weekend vs weekday sales) that run against your star schema using simple JOINs.
  • A side-by-side comparison showing how the same business questions go from ugly, verbose SQL against flat data to elegant, readable queries against your dimensional model.
  • Secret Mission: Add data quality validation with referential integrity checks and Kimball placeholder rows for handling missing dimension keys.

Are there any prerequisites?

You should have basic Python scripting and pandas experience (creating DataFrames, reading CSVs).

Basic SQL familiarity is also helpful. You should know SELECT, INSERT, and CREATE TABLE statements.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Dimensional modeling and ETL pipelines are two of the most commonly listed skills on data engineering job descriptions. This project gives you hands-on practice with both.

Set Up PostgreSQL and Your Python Environment

Your ETL pipeline needs a database to serve as its data warehouse. PostgreSQL is the relational database engine you'll use to store both raw staging data and the finished star schema.

In this step, you'll install PostgreSQL 18 from the official package repository and confirm your Python environment can connect to it.

In this step, get ready to:
  • Install PostgreSQL 18 from the official PGDG apt repository.
  • Set the postgres user password so your Python scripts can authenticate.
  • Verify both PostgreSQL and your Python libraries are working together.
Install PostgreSQL 18 using the PGDG apt repository

Ubuntu's default repositories often ship an older PostgreSQL version. The PGDG (PostgreSQL Global Development Group) apt repository provides the latest stable release.

  • Install the postgresql-common package and certificate utilities by running this command:
sudo apt install -y postgresql-common ca-certificates
  • Configure the PGDG apt repository by running the setup script:
sudo /usr/share/postgresql-common/pgdg/apt.postgresql.org.sh

What does this script do?

This script automatically detects your Ubuntu version and adds the official PostgreSQL package repository to your system. It also imports the signing key so apt trusts packages from this repository.

The script will ask you to confirm before making changes. Press Enter or type yes when prompted.

  • Update your package list and install PostgreSQL 18 by running these commands:
sudo apt update
sudo apt install -y postgresql-18

The installation takes a minute or two. Once it finishes, PostgreSQL automatically starts as a background service.

  • Confirm the service is running by checking its status:
sudo systemctl status postgresql

You should see active (exited) or active (running) in the output, confirming PostgreSQL is live.

Service not active?

If the status shows inactive or failed, start it manually with sudo systemctl start postgresql and then re-check the status.

Help me troubleshoot why PostgreSQL won't start

Set the postgres user password

PostgreSQL creates a system user called postgres with no password by default. Your Python scripts need password-based authentication to connect, so you'll set one now.

  • Set the password to postgres by running this command:
sudo -u postgres psql -c "ALTER USER postgres PASSWORD 'postgres';"

What does this command do?

The sudo -u postgres part runs the command as the postgres system user, which has local trust access to the database.

The ALTER USER SQL command sets a password so that TCP connections (the kind your Python scripts use) can authenticate.

You should see ALTER ROLE printed, confirming the password was set.

Verify PostgreSQL and Python libraries are working

You already have pandas and psycopg2-binary installed in your virtual environment from earlier. Now you'll confirm that everything connects end to end.

  • Before you run the next command, what version number do you expect to see in the output?
  • Test the PostgreSQL connection by running:
psql -U postgres -h localhost -c "SELECT version();"

When prompted for a password, type postgres (the password you just set).

✔️ I see PostgreSQL 18.x in the output

Your PostgreSQL installation is working. The output shows the full version string confirming you're running version 18.

ⓧ I see an older PostgreSQL version

You may have a previous PostgreSQL installation that is responding instead of the new one.

  • Check which clusters are installed by running:
pg_lsclusters

This shows all PostgreSQL clusters with their version, port, and status.

  • If you see an older version's cluster using port 5432, drop it and restart PostgreSQL:
sudo pg_dropcluster --stop <old_version> main
sudo systemctl restart postgresql
  • Replace <old_version> with the version number shown by pg_lsclusters (for example, 16 or 17).
  • Re-run the version check command and confirm it now shows version 18.

ⓧ Connection refused or command not found

This means either the service is not running or psql is not in your PATH.

  • Make sure PostgreSQL is running:
sudo systemctl start postgresql
  • If psql is not found, install the client tools:
sudo apt install -y postgresql-client-18
  • Try the version check command again.

Help me fix my PostgreSQL connection issue

  • Now verify your Python libraries by running:
python -c "import pandas; import psycopg2; print('Ready')"

You should see Ready printed with no errors.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. You should see the environment name in parentheses at the start of your terminal prompt.

If it is not active, activate it and try the import command again.

Help me fix my Python import error

Your database is running and your Python environment can talk to it. Next up, you'll generate sample sales data and load it into a staging table so you have something real to transform.

Generate Sample Data and Load It Into Staging

Your PostgreSQL database is running and your Python environment is ready. Now you need something to actually work with.

In this step, you will generate realistic retail sales data and load it into a staging table. This is how data arrives in the real world: flat, denormalized, and repetitive. Every single row repeats the customer's name, email, city, and product details. You will feel that pain firsthand in the next step.

In this step, get ready to:
  • Create a Python script that generates 500 sample sales records as a CSV file.
  • Create a database setup script that builds the staging table and bulk-loads the CSV into PostgreSQL.
Create the data generation script

Before you can build a data warehouse, you need source data. You will write a script that simulates what a retail point-of-sale system might export: a flat CSV where every order row contains all customer and product details repeated inline.

  • In VS Code, create a new file called generate_data.py in your project folder.
  • Add the imports, random seed, and data definitions by pasting this code:
import pandas as pd
import random
from datetime import datetime, timedelta

random.seed(42)

customers = [
    ("Alice Johnson", "alice.johnson@email.com", "Vancouver", "BC"),
    ("Bob Smith", "bob.smith@email.com", "Toronto", "ON"),
    ("Carol Williams", "carol.williams@email.com", "Montreal", "QC"),
    ("David Brown", "david.brown@email.com", "Calgary", "AB"),
    ("Eva Martinez", "eva.martinez@email.com", "Ottawa", "ON"),
    ("Frank Lee", "frank.lee@email.com", "Edmonton", "AB"),
    ("Grace Kim", "grace.kim@email.com", "Vancouver", "BC"),
    ("Henry Wilson", "henry.wilson@email.com", "Victoria", "BC"),
]

products = [
    ("Laptop Pro 15", "Electronics", "Computers", 1299.99),
    ("Wireless Mouse", "Electronics", "Accessories", 29.99),
    ("USB-C Hub", "Electronics", "Accessories", 49.99),
    ("Standing Desk", "Furniture", "Desks", 599.99),
    ("Ergonomic Chair", "Furniture", "Chairs", 449.99),
    ("Monitor 27 inch", "Electronics", "Displays", 399.99),
    ("Keyboard Mechanical", "Electronics", "Accessories", 89.99),
    ("Desk Lamp LED", "Furniture", "Lighting", 39.99),
    ("Webcam HD", "Electronics", "Accessories", 79.99),
    ("Filing Cabinet", "Furniture", "Storage", 199.99),
]

What does this code do?

This sets up the building blocks for your sample data. random.seed(42) ensures everyone who runs this script gets the same "random" data, which makes debugging easier.

The customers list holds 8 people with name, email, city, and province. The products list holds 10 items with name, category, subcategory, and price. Each order will randomly pick from these lists.

  • Below the products list, add the generation loop that builds 500 order records:
discount_options = [0.0, 0.0, 0.0, 0.05, 0.10, 0.15, 0.20]

start_date = datetime(2024, 1, 1)
end_date = datetime(2024, 6, 30)
date_range = (end_date - start_date).days

records = []
for i in range(1, 501):
    order_date = start_date + timedelta(days=random.randint(0, date_range))
    customer = random.choice(customers)
    product = random.choice(products)
    quantity = random.randint(1, 5)
    discount_pct = random.choice(discount_options)

    records.append({
        "order_id": f"ORD-{i:04d}",
        "order_date": order_date.strftime("%Y-%m-%d"),
        "customer_name": customer[0],
        "customer_email": customer[1],
        "customer_city": customer[2],
        "customer_state": customer[3],
        "product_name": product[0],
        "product_category": product[1],
        "product_subcategory": product[2],
        "unit_price": product[3],
        "quantity": quantity,
        "discount_pct": discount_pct,
    })

What does this code do?

The discount_options list weights discounts so most orders have no discount (three 0.0 entries), mimicking real retail patterns.

The loop creates 500 orders spread across January to June 2024. Each row packs the full customer tuple and full product tuple into a flat dictionary. This is the "denormalized" shape you will fix later with your star schema.

  • Complete the script by adding the output lines below the loop:
df = pd.DataFrame(records)
df.to_csv("raw_sales.csv", index=False)
print(f"Generated {len(df)} sales records in raw_sales.csv")
print(df.head())

What does this code do?

These lines convert the list of dictionaries into a pandas DataFrame, write it out to raw_sales.csv, and print a confirmation with a preview of the first 5 rows.

  • Save generate_data.py.
  • Run the script by executing this command in your terminal:
python generate_data.py

You should see output showing 500 records were generated, followed by the first 5 rows of data.

  • Confirm the CSV file was created by running:
ls raw_sales.csv

You should see raw_sales.csv listed in the output.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. You should see the environment name in your terminal prompt. If not, activate it again with source venv/bin/activate and try running the script again.

Help me fix this ModuleNotFoundError in my data generation script.

✔️ Awesome, I've got everything!

Great. Double check you have saved generate_data.py and that raw_sales.csv exists in the same folder.

ⓧ I'd like to double check the full code

Your generate_data.py should contain these three sections in order:

  • The imports, random.seed(42), the customers list (8 entries), and the products list (10 entries).
  • The discount_options list, date range setup, and the for i in range(1, 501) loop that builds records with records.append({...}).
  • The df = pd.DataFrame(records) line, df.to_csv("raw_sales.csv", index=False), and the two print statements.

Scroll up and confirm all three sections are present in your file, in order.

Create the database setup script

Now you need somewhere to put this data inside PostgreSQL. A staging layer is a holding area where raw data lands before any transformation happens. You will create a dedicated database, a staging schema inside it, and a table that mirrors your CSV columns exactly.

  • Create a new file called setup_database.py in the same folder as generate_data.py.
  • Add the import and connection configuration:
import psycopg2

DB_CONFIG = {
    "dbname": "retail_dw",
    "user": "postgres",
    "password": "postgres",
    "host": "localhost",
    "port": 5432,
}

What is DB_CONFIG?

This dictionary holds the connection details for your data warehouse database. Every function in this script unpacks it with **DB_CONFIG to connect. The database name retail_dw stands for "retail data warehouse."

  • Below the config, add the function that creates the database:
def create_database():
    conn = psycopg2.connect(
        dbname="postgres",
        user=DB_CONFIG["user"],
        password=DB_CONFIG["password"],
        host=DB_CONFIG["host"],
        port=DB_CONFIG["port"],
    )
    conn.autocommit = True
    cursor = conn.cursor()

    cursor.execute(
        "SELECT 1 FROM pg_database WHERE datname = %s", ("retail_dw",)
    )
    if not cursor.fetchone():
        cursor.execute("CREATE DATABASE retail_dw")
        print("Database 'retail_dw' created.")
    else:
        print("Database 'retail_dw' already exists.")

    cursor.close()
    conn.close()

Why connect to the postgres database first?

You cannot create a new database while connected to it. The function connects to the default postgres database, checks if retail_dw already exists (making the script safe to run repeatedly), and only creates it if missing.

Setting autocommit = True is required because CREATE DATABASE cannot run inside a transaction.

  • Save setup_database.py and verify the database gets created by running:
python -c "from setup_database import create_database; create_database()"

You should see Database 'retail_dw' created. printed in your terminal.

Seeing a connection refused error?

Confirm PostgreSQL is running with sudo systemctl status postgresql. If it shows inactive, start it with sudo systemctl start postgresql.

Also verify the password is correct. If you set a different password for the postgres user in Step 1, update the password value in DB_CONFIG to match.

Help me fix this PostgreSQL connection error in my setup script.

  • Now add the function that creates the staging table. Paste this below create_database:
def create_staging_table():
    conn = psycopg2.connect(**DB_CONFIG)
    cursor = conn.cursor()

    cursor.execute("CREATE SCHEMA IF NOT EXISTS staging")

    cursor.execute("""
        DROP TABLE IF EXISTS staging.raw_sales;
        CREATE TABLE staging.raw_sales (
            order_id VARCHAR(20),
            order_date DATE,
            customer_name VARCHAR(100),
            customer_email VARCHAR(100),
            customer_city VARCHAR(50),
            customer_state VARCHAR(10),
            product_name VARCHAR(100),
            product_category VARCHAR(50),
            product_subcategory VARCHAR(50),
            unit_price NUMERIC(10, 2),
            quantity INTEGER,
            discount_pct NUMERIC(5, 2)
        )
    """)

    conn.commit()
    print("Staging table 'staging.raw_sales' created.")
    cursor.close()
    conn.close()

What is a staging schema?

A schema in PostgreSQL is like a folder inside your database. Putting raw data in a staging schema keeps it separate from your cleaned warehouse data. The table mirrors your CSV exactly: every column from the flat file gets its own column here.

  • Finally, add the load function and the main block that ties everything together:
def load_csv_to_staging():
    conn = psycopg2.connect(**DB_CONFIG)
    cursor = conn.cursor()

    with open("raw_sales.csv", "r") as f:
        cursor.copy_expert(
            "COPY staging.raw_sales FROM STDIN WITH CSV HEADER", f
        )

    conn.commit()

    cursor.execute("SELECT COUNT(*) FROM staging.raw_sales")
    count = cursor.fetchone()[0]
    print(f"Loaded {count} rows into staging.raw_sales")

    cursor.close()
    conn.close()


if __name__ == "__main__":
    create_database()
    create_staging_table()
    load_csv_to_staging()

What is copy_expert?

PostgreSQL's COPY command is the fastest way to bulk-load data from a file. The psycopg2 method copy_expert streams your CSV directly to PostgreSQL without loading it all into Python memory first. For 500 rows this is overkill, but in production you would use this same pattern to load millions of rows.

The if __name__ == "__main__" block calls all three functions in order: create database, create table, then load data.

  • Save setup_database.py.
  • Before you run the full script, what do you expect to see printed? The script calls three functions in sequence.
  • Run the full setup by executing:
python setup_database.py

You should see three lines confirming each phase completed. The database already exists from your earlier test, so the first line reports that. The second line confirms the staging table was created. The third line shows 500 rows loaded.

Seeing a FileNotFoundError for raw_sales.csv?

Make sure you are running the command from the same folder that contains raw_sales.csv. Run ls raw_sales.csv to confirm it exists in your current directory.

Help me fix this FileNotFoundError in my database setup script.

  • Verify the data landed in PostgreSQL by running this query directly:
psql -U postgres -h localhost -d retail_dw -c "SELECT * FROM staging.raw_sales LIMIT 5;"

You should see five rows of flat sales data with all 12 columns displayed. Notice how customer_name, customer_email, and customer_city repeat on every single row. That duplication is what you will eliminate with the star schema.

Prompted for a password?

Enter postgres when prompted. This is the password you set for the postgres user in Step 1.

Help me connect to my PostgreSQL database with psql.

✔️ Awesome, I've got everything!

Great. Double check you have saved setup_database.py and that all three functions ran successfully.

ⓧ I'd like to double check the full code

Your setup_database.py should contain these sections in order:

  • The import psycopg2 line and the DB_CONFIG dictionary with dbname, user, password, host, and port.
  • The create_database() function that connects to the default postgres database and creates retail_dw if it does not exist.
  • The create_staging_table() function that creates the staging schema and the raw_sales table with 12 columns.
  • The load_csv_to_staging() function using copy_expert, followed by the if __name__ == "__main__" block calling all three functions.

Scroll up and confirm each section is present in your file, in order.

You now have 500 rows of messy, flat retail data sitting in your staging table. Next, you will try to answer business questions against this flat data and discover exactly why dimensional modeling exists.

Query the Flat Data and Feel the Pain

You have 500 rows of sales data in your staging table. Time to put it to work by answering real business questions: what is the monthly revenue by product category, and who are the top customers?

In this step, you will try to answer those questions directly against the flat staging data. The experience of writing those queries will reveal exactly why data engineers reach for dimensional modeling.

In this step, get ready to:
  • Query monthly revenue by product category and top customers against the flat staging table.
  • Measure how much customer data is duplicated across rows.
Ask the staging table for monthly revenue by category

Imagine your manager asks: "What was our monthly revenue broken down by product category?" You have all the data in one flat table, so this should be straightforward.

  • Connect to the retail_dw database in psql by running this command:
psql -U postgres -h localhost -d retail_dw

Before you run the next query, think about this: how many columns do you think you will need to list in the GROUP BY clause to answer "monthly revenue by product category"?

  • Run this query to get monthly revenue by product category:
SELECT
    DATE_TRUNC('month', order_date) AS month,
    product_category,
    product_subcategory,
    SUM(unit_price * quantity * (1 - discount_pct)) AS revenue,
    COUNT(*) AS total_orders
FROM staging.raw_sales
GROUP BY
    DATE_TRUNC('month', order_date),
    product_category,
    product_subcategory
ORDER BY month, revenue DESC;

You should see a table of results with monthly revenue broken down by category and subcategory. It works, but look at that GROUP BY clause.

What's wrong with this query?

The query requires you to repeat every non-aggregated column in the GROUP BY clause. You listed product_category and product_subcategory in both SELECT and GROUP BY.

With a flat table, every time you add a column to your analysis, you must also add it to GROUP BY. This gets verbose fast.

Now try the second question: who are the top 5 customers by revenue?

  • Run this query to find the top 5 customers:
SELECT
    customer_name,
    customer_email,
    customer_city,
    customer_state,
    COUNT(*) AS total_orders,
    SUM(unit_price * quantity * (1 - discount_pct)) AS total_revenue
FROM staging.raw_sales
GROUP BY
    customer_name,
    customer_email,
    customer_city,
    customer_state
ORDER BY total_revenue DESC
LIMIT 5;

You should see five customer rows with their order counts and revenue. Notice that you had to list four separate columns in GROUP BY just to identify one customer.

Why is this painful?

A customer is a single entity, but the flat table has no concept of "a customer." You must manually group by every column that describes a customer: name, email, city, and state. Miss one and you get incorrect groupings.

In a dimensional model, you would simply JOIN to a customer dimension table and group by a single customer_key. The difference becomes dramatic as queries grow more complex.

Count the duplication

The verbose GROUP BY is only half the problem. The other half is that customer details are physically copied into every single row for that customer. Let's see how bad the duplication really is.

  • Run this query to count how many times each customer's full details are stored:
SELECT
    customer_name,
    customer_email,
    customer_city,
    COUNT(*) AS times_stored
FROM staging.raw_sales
GROUP BY customer_name, customer_email, customer_city
ORDER BY times_stored DESC;

You should see each of the 8 customers listed with the number of rows their details appear in. Every customer's name, email, and city are stored 50+ times.

What does this duplication cost you?

With 8 customers and 500 rows, each customer's details are repeated roughly 60 times on average. In a normalized dimension table, each customer would be stored exactly once.

At production scale with millions of rows, this duplication wastes storage and makes updates dangerous. If Alice moves to a new city, you would need to update hundreds of rows instead of one.

  • Run this query to see the total row count versus unique customers:
SELECT
    COUNT(*) AS total_rows,
    COUNT(DISTINCT customer_email) AS unique_customers
FROM staging.raw_sales;

You should see 500 total rows but only 8 unique customers. That is 500 copies of customer data to represent 8 people.

  • Exit psql by running:
\q

Queries returning errors?

Make sure you connected to the retail_dw database, not the default postgres database. The staging schema only exists in retail_dw.

If you see "relation staging.raw_sales does not exist", reconnect with the correct database name in your psql command.

Help me debug my psql query against staging.raw_sales.

Now you have felt the pain firsthand. Verbose GROUP BY clauses, repeated column references, and massive data duplication across every row. In the next step, you will design a star schema that eliminates all three problems.

Design and Create the Star Schema

In the last step, you experienced firsthand why flat data is painful. Answering simple business questions required verbose GROUP BY clauses, and customer details were duplicated across hundreds of rows.

The star schema solves these problems. Dimension tables store each entity exactly once (eliminating duplication), surrogate keys make joins fast and stable, and the fact table holds only measurements and foreign keys. In this step, you will design and create this structure in your database.

In this step, get ready to:
  • Understand the Kimball star schema design with four tables.
  • Create a Python script that builds the star schema in PostgreSQL.
  • Verify the warehouse tables exist in the database.
Understand the Kimball star schema design

The Kimball dimensional modeling methodology follows four steps: select the business process (retail sales), declare the grain (one row per order line item), identify dimensions (date, customer, product), and identify facts (quantity, price, discount, net amount).

Your star schema will have four tables. Three dimension tables hold descriptive attributes, and one fact table holds numeric measurements connected to the dimensions by foreign keys.

What are the four tables?

  • dim_date stores calendar attributes for each order date. Its primary key is an integer in YYYYMMDD format (like 20240115 for January 15, 2024). This is a common Kimball pattern called a smart key because the key itself carries meaning.
  • dim_customer stores each customer exactly once. It uses a SERIAL surrogate key that auto-increments. The natural attributes (name, email, city, state) describe the customer.
  • dim_product stores each product once with its category hierarchy. It also uses a SERIAL surrogate key.
  • fact_sales holds one row per order. It contains foreign keys to all three dimensions plus numeric measures: quantity, unit_price, discount_pct, discount_amount, and net_amount.

Why surrogate keys instead of natural keys?

A surrogate key is a database-generated integer that has no business meaning. It shields the warehouse from source-system changes. If a customer's email changes, their surrogate key stays the same, so all historical fact rows still link correctly.

In PostgreSQL, SERIAL PRIMARY KEY creates an auto-incrementing integer column backed by a sequence. Each time you insert a row without specifying the key, PostgreSQL assigns the next number automatically.

Create the star schema script
  • In VS Code, create a new file called create_star_schema.py in the same folder as your other scripts.
  • Paste the following code to set up the database connection, create the warehouse schema, and define the DROP statements that ensure a clean slate:
import psycopg2

DB_CONFIG = {
    "dbname": "retail_dw",
    "user": "postgres",
    "password": "postgres",
    "host": "localhost",
    "port": 5432,
}


def create_warehouse_schema():
    conn = psycopg2.connect(**DB_CONFIG)
    cursor = conn.cursor()

    cursor.execute("CREATE SCHEMA IF NOT EXISTS warehouse")

    cursor.execute("""
        DROP TABLE IF EXISTS warehouse.fact_sales;
        DROP TABLE IF EXISTS warehouse.dim_date;
        DROP TABLE IF EXISTS warehouse.dim_customer;
        DROP TABLE IF EXISTS warehouse.dim_product;

What does this code do?

  • DB_CONFIG reuses the same connection settings as your setup_database.py script to connect to the retail_dw database.
  • CREATE SCHEMA IF NOT EXISTS warehouse creates a separate namespace for the star schema tables, keeping them isolated from the staging data.
  • The DROP statements remove existing tables in dependency order (fact first, then dimensions) so you can rerun the script without errors.
  • Save create_star_schema.py.
  • Confirm the file appears in your VS Code Explorer sidebar alongside generate_data.py and setup_database.py.
  • Now add the three dimension table definitions directly below the code you just pasted (still inside the triple-quoted SQL string):

        CREATE TABLE warehouse.dim_date (
            date_key INTEGER PRIMARY KEY,
            full_date DATE NOT NULL,
            day_of_week INTEGER NOT NULL,
            day_name VARCHAR(10) NOT NULL,
            month INTEGER NOT NULL,
            month_name VARCHAR(10) NOT NULL,
            quarter INTEGER NOT NULL,
            year INTEGER NOT NULL,
            is_weekend BOOLEAN NOT NULL
        );

        CREATE TABLE warehouse.dim_customer (
            customer_key SERIAL PRIMARY KEY,
            customer_name VARCHAR(100) NOT NULL,
            customer_email VARCHAR(100) NOT NULL,
            customer_city VARCHAR(50) NOT NULL,
            customer_state VARCHAR(10) NOT NULL
        );

        CREATE TABLE warehouse.dim_product (
            product_key SERIAL PRIMARY KEY,
            product_name VARCHAR(100) NOT NULL,
            product_category VARCHAR(50) NOT NULL,
            product_subcategory VARCHAR(50) NOT NULL,
            unit_price NUMERIC(10, 2) NOT NULL
        );

What do these dimension tables store?

  • dim_date uses date_key INTEGER PRIMARY KEY in YYYYMMDD format. This lets your ETL pipeline generate the key from the date itself without needing a lookup table.
  • dim_customer and dim_product use SERIAL PRIMARY KEY for auto-generated surrogate keys. Each customer and product is stored exactly once, eliminating the duplication you saw in the flat staging table.
  • Every column has NOT NULL constraints to enforce data quality at the database level.
  • Add the fact table definition and the closing portion of the script directly below the dimension tables:

        CREATE TABLE warehouse.fact_sales (
            sale_key SERIAL PRIMARY KEY,
            date_key INTEGER NOT NULL REFERENCES warehouse.dim_date(date_key),
            customer_key INTEGER NOT NULL REFERENCES warehouse.dim_customer(customer_key),
            product_key INTEGER NOT NULL REFERENCES warehouse.dim_product(product_key),
            order_id VARCHAR(20) NOT NULL,
            quantity INTEGER NOT NULL,
            unit_price NUMERIC(10, 2) NOT NULL,
            discount_pct NUMERIC(5, 2) NOT NULL,
            discount_amount NUMERIC(10, 2) NOT NULL,
            net_amount NUMERIC(10, 2) NOT NULL
        );
    """)

    conn.commit()
    print("Star schema created in 'warehouse' schema.")
    print("  - warehouse.dim_date")
    print("  - warehouse.dim_customer")
    print("  - warehouse.dim_product")
    print("  - warehouse.fact_sales")

    cursor.close()
    conn.close()


if __name__ == "__main__":
    create_warehouse_schema()

What does the fact table do?

  • REFERENCES creates foreign key constraints linking each fact row to its dimension tables. PostgreSQL will reject any insert that references a non-existent dimension key.
  • The measures (quantity, unit_price, discount_pct, discount_amount, net_amount) are the numeric values you will aggregate in analytical queries.
  • conn.commit() saves all DDL changes to the database. Without this, the tables would not persist.
  • Save create_star_schema.py.

✔️ Awesome, I've got everything!

Double check you have saved your file. The complete script should start with import psycopg2 and end with create_warehouse_schema().

ⓧ I'd like to double check the full code

Your create_star_schema.py should contain these three sections in order:

  • The connection setup chunk: import psycopg2, DB_CONFIG, the create_warehouse_schema() function definition, CREATE SCHEMA, and the four DROP TABLE statements.
  • The dimension table definitions: CREATE TABLE warehouse.dim_date, CREATE TABLE warehouse.dim_customer, and CREATE TABLE warehouse.dim_product.
  • The fact table and closing code: CREATE TABLE warehouse.fact_sales, the closing """), conn.commit(), the print statements, cursor.close(), conn.close(), and the if __name__ block.

Scroll up and confirm each section matches what you pasted.

Run the script and verify the tables

Before you run this, what do you expect to see printed in your terminal? Think about what the script does at the end of the function.

  • Run the schema creation script in your terminal:
python create_star_schema.py

You should see confirmation that all four tables were created:

Star schema created in 'warehouse' schema.
  - warehouse.dim_date
  - warehouse.dim_customer
  - warehouse.dim_product
  - warehouse.fact_sales
  • Verify the tables exist in PostgreSQL by running:
psql -U postgres -h localhost -d retail_dw -c "\dt warehouse.*"

You should see all four tables listed under the warehouse schema: dim_date, dim_customer, dim_product, and fact_sales.

Not seeing the tables?

  • If you get a connection error, make sure PostgreSQL is running with sudo systemctl status postgresql.
  • If you see a syntax error in Python, check that the triple-quoted SQL string opens with cursor.execute(""" and closes with """) on its own line.
  • If \dt warehouse.* shows nothing, confirm the script printed its success messages. If it did not, check for a password mismatch in DB_CONFIG.

Help me debug my create_star_schema.py script.

Your star schema is in place. The warehouse has a clean structure with dimension tables for each entity and a fact table ready to receive measurements. Next up, you will build the ETL pipeline that transforms your flat staging data and loads it into these tables.

Build and Run the ETL Pipeline

Your star schema is in place, but every table in the warehouse schema is empty. The schema is just a blueprint.

The real data engineering work is building the pipeline that moves data from staging into that blueprint. In this step, you will build a Python ETL pipeline that extracts raw data, transforms flat rows into proper dimensions and facts, then loads everything into the star schema.

In this step, get ready to:
  • Write extract and transform functions that reshape flat data into dimensions and facts.
  • Add load functions that insert data into the star schema respecting foreign key order.
  • Run analytical queries that prove the star schema makes business questions easy to answer.
Create the extract and transform functions

The ETL pipeline lives in a single file with functions for each phase. You will start with the Extract phase (pulling data from staging) and the Transform phase (reshaping flat rows into dimension and fact DataFrames).

  • In VS Code, create a new file called etl_pipeline.py in your project folder.
  • Add the imports, database config, and the extract function by pasting this code:
import pandas as pd
import psycopg2

DB_CONFIG = {
    "dbname": "retail_dw",
    "user": "postgres",
    "password": "postgres",
    "host": "localhost",
    "port": 5432,
}


def extract_staging_data():
    # Pull all raw sales data from the staging table
    conn = psycopg2.connect(**DB_CONFIG)
    df = pd.read_sql("SELECT * FROM staging.raw_sales", conn)
    conn.close()
    print(f"Extracted {len(df)} rows from staging.")
    return df

What does this code do?

The extract_staging_data function is the E in ETL. It connects to PostgreSQL, runs a SELECT query to pull all 500 rows from staging.raw_sales, and returns them as a pandas DataFrame.

This is the Extract phase: grabbing raw data from the source system exactly as it is, before any transformation.

  • Save etl_pipeline.py.
  • Verify the extract function works by running this command:
python -c "from etl_pipeline import extract_staging_data; extract_staging_data()"

You should see Extracted 500 rows from staging. printed in your terminal.

Seeing a warning about SQLAlchemy?

You may see a warning mentioning that pandas only supports SQLAlchemy connectable objects. This is non-blocking. The function still works correctly with a raw psycopg2 connection and you can safely ignore it.

If you see a connection error instead, confirm PostgreSQL is running and your password matches the DB_CONFIG dictionary. Help me fix my database connection error in etl_pipeline.py

Now add the first transform function. This builds the date dimension by extracting unique dates from the raw data and enriching them with calendar attributes.

  • Add this function below extract_staging_data in etl_pipeline.py:
def transform_dim_date(df):
    # Extract unique dates and build calendar attributes
    unique_dates = pd.to_datetime(df["order_date"]).dt.date.unique()

    date_records = []
    for d in sorted(unique_dates):
        date_records.append({
            "date_key": int(d.strftime("%Y%m%d")),
            "full_date": d,
            "day_of_week": d.isoweekday(),
            "day_name": d.strftime("%A"),
            "month": d.month,
            "month_name": d.strftime("%B"),
            "quarter": (d.month - 1) // 3 + 1,
            "year": d.year,
            "is_weekend": d.isoweekday() in (6, 7),
        })

    return pd.DataFrame(date_records)

What does this code do?

This function builds dim_date by finding every unique order date and creating a row with calendar attributes for each one.

The date_key is an integer in YYYYMMDD format (e.g. 20240115 for January 15, 2024). This is a common Kimball convention that makes date-based filtering fast and human-readable.

  • Next, add the customer and product dimension transforms below transform_dim_date:
def transform_dim_customer(df):
    # Deduplicate customers and assign surrogate keys
    dim_customer = df[
        ["customer_name", "customer_email", "customer_city", "customer_state"]
    ].drop_duplicates().reset_index(drop=True)

    dim_customer.index += 1
    dim_customer.index.name = "customer_key"

    return dim_customer.reset_index()


def transform_dim_product(df):
    # Deduplicate products and assign surrogate keys
    dim_product = df[
        ["product_name", "product_category", "product_subcategory", "unit_price"]
    ].drop_duplicates().reset_index(drop=True)

    dim_product.index += 1
    dim_product.index.name = "product_key"

    return dim_product.reset_index()

What are surrogate keys?

A surrogate key is a simple integer that uniquely identifies each dimension row. It replaces the natural key (like email or product name) as the join key in the fact table.

This is a core Kimball technique. If a customer changes their email, the surrogate key stays the same. Your fact table joins remain stable regardless of source system changes.

  • Save etl_pipeline.py and verify the transforms produce the right counts by running:
python -c "from etl_pipeline import *; df = extract_staging_data(); print(len(transform_dim_customer(df)), 'customers'); print(len(transform_dim_product(df)), 'products')"

You should see 8 customers and 10 products printed. These match the 8 unique customers and 10 unique products in your source data.

The final transform is the most complex. It builds the fact table by looking up surrogate keys from each dimension and calculating new measures.

  • Add the fact table transform below transform_dim_product:
def transform_fact_sales(df, dim_date, dim_customer, dim_product):
    fact = df.copy()

    # Convert order_date to YYYYMMDD integer for date_key lookup
    fact["date_key"] = pd.to_datetime(fact["order_date"]).dt.strftime("%Y%m%d").astype(int)

    # Look up surrogate keys from dimension tables
    customer_lookup = dim_customer.set_index("customer_email")["customer_key"]
    fact["customer_key"] = fact["customer_email"].map(customer_lookup)

    product_lookup = dim_product.set_index("product_name")["product_key"]
    fact["product_key"] = fact["product_name"].map(product_lookup)

    # Calculate derived measures
    fact["discount_amount"] = (
        fact["unit_price"] * fact["quantity"] * fact["discount_pct"]
    ).round(2)
    fact["net_amount"] = (
        fact["unit_price"] * fact["quantity"] - fact["discount_amount"]
    ).round(2)

    fact_sales = fact[
        ["date_key", "customer_key", "product_key", "order_id",
         "quantity", "unit_price", "discount_pct", "discount_amount", "net_amount"]
    ]

    return fact_sales

What does this code do?

  • date_key is computed by formatting the order date as YYYYMMDD, matching the keys in dim_date.
  • customer_key and product_key are looked up by mapping natural identifiers (email, product name) to their surrogate keys in the dimension DataFrames.
  • discount_amount and net_amount are calculated measures. The fact table only stores foreign keys and numeric measurements.
Add the load phase and run the pipeline

The Transform phase produces DataFrames ready to load. Now you need functions that INSERT those rows into PostgreSQL, plus an orchestrator function that ties Extract, Transform, and Load together.

  • Add the dimension loader function below transform_fact_sales in etl_pipeline.py:
def load_dimension(conn, table_name, df, columns):
    # Insert each dimension row into the target table
    cursor = conn.cursor()
    for _, row in df.iterrows():
        values = [row[col] for col in columns]
        placeholders = ", ".join(["%s"] * len(columns))
        col_names = ", ".join(columns)
        cursor.execute(
            f"INSERT INTO {table_name} ({col_names}) VALUES ({placeholders})",
            values,
        )
    conn.commit()
    cursor.close()
    print(f"  Loaded {len(df)} rows into {table_name}")

What does this code do?

The load_dimension function is generic. It accepts any table name and column list, so you can reuse it for all three dimensions. It iterates over each DataFrame row and inserts it with parameterized SQL to prevent injection.

  • Add the fact table loader directly below load_dimension:
def load_fact(conn, df):
    # Insert each fact row into the fact_sales table
    cursor = conn.cursor()
    columns = [
        "date_key", "customer_key", "product_key", "order_id",
        "quantity", "unit_price", "discount_pct", "discount_amount", "net_amount",
    ]
    for _, row in df.iterrows():
        values = [row[col] for col in columns]
        placeholders = ", ".join(["%s"] * len(columns))
        col_names = ", ".join(columns)
        cursor.execute(
            f"INSERT INTO warehouse.fact_sales ({col_names}) VALUES ({placeholders})",
            values,
        )
    conn.commit()
    cursor.close()
    print(f"  Loaded {len(df)} rows into warehouse.fact_sales")

Why load dimensions before facts?

The fact table has foreign key constraints pointing to each dimension table. If you tried to insert a fact row before the matching dimension row exists, PostgreSQL would reject it.

The load_fact function is specific to warehouse.fact_sales because the fact table has a fixed column set that differs from the generic dimension pattern.

  • Save etl_pipeline.py and confirm the file has no syntax errors by running:
python -c "import etl_pipeline; print('All functions defined')"

You should see All functions defined printed with no errors.

Now add the orchestrator function that calls Extract, Transform, and Load in sequence.

  • Add the run_etl function below load_fact:
def run_etl():
    print("=" * 50)
    print("ETL Pipeline: Staging -> Star Schema")
    print("=" * 50)

    print("\n[1/3] EXTRACT")
    raw_data = extract_staging_data()

    print("\n[2/3] TRANSFORM")
    dim_date = transform_dim_date(raw_data)
    print(f"  dim_date: {len(dim_date)} rows")

    dim_customer = transform_dim_customer(raw_data)
    print(f"  dim_customer: {len(dim_customer)} rows")

    dim_product = transform_dim_product(raw_data)
    print(f"  dim_product: {len(dim_product)} rows")

    fact_sales = transform_fact_sales(raw_data, dim_date, dim_customer, dim_product)
    print(f"  fact_sales: {len(fact_sales)} rows")

What does this code do?

This is the first half of your orchestrator. It extracts the raw data, then calls each transform function in order, printing row counts so you can track the pipeline's progress.

Notice that transform_fact_sales receives all three dimension DataFrames as arguments. It needs them to look up surrogate keys.

  • Continue the run_etl function by adding the Load phase directly below the last line you pasted (still indented inside the function):
    print("\n[3/3] LOAD")
    conn = psycopg2.connect(**DB_CONFIG)

    load_dimension(
        conn, "warehouse.dim_date", dim_date,
        ["date_key", "full_date", "day_of_week", "day_name",
         "month", "month_name", "quarter", "year", "is_weekend"],
    )
    load_dimension(
        conn, "warehouse.dim_customer", dim_customer,
        ["customer_key", "customer_name", "customer_email",
         "customer_city", "customer_state"],
    )
    load_dimension(
        conn, "warehouse.dim_product", dim_product,
        ["product_key", "product_name", "product_category",
         "product_subcategory", "unit_price"],
    )
    load_fact(conn, fact_sales)

    conn.close()
    print("\nETL pipeline complete!")


if __name__ == "__main__":
    run_etl()

What does this code do?

The Load phase opens a single database connection and inserts dimensions first (date, customer, product), then the fact table. This order satisfies the foreign key constraints.

The if __name__ == "__main__" block means the pipeline runs when you execute the file directly from your terminal.

  • Save etl_pipeline.py.

Before you run the pipeline, what do you expect to see? Think about how many rows should load into each table based on the 8 customers, 10 products, and 500 orders in your staging data.

  • Run the full ETL pipeline:
python etl_pipeline.py

You should see output showing all three phases completing. The Extract phase pulls 500 rows. The Transform phase shows row counts for each dimension and the fact table. The Load phase confirms rows inserted into each warehouse table, ending with ETL pipeline complete!.

Seeing a foreign key violation error?

If you see an error about violating a foreign key constraint, the most likely cause is that run_etl is loading the fact table before a dimension. Check that your Load section calls load_dimension for all three dimensions BEFORE load_fact.

If you already ran the pipeline once and are running it again, the tables already have data. Re-run python create_star_schema.py first to drop and recreate the empty tables, then try the pipeline again. Help me fix my ETL pipeline foreign key error

✔️ Awesome, I've got everything!

Great. Double check you have saved etl_pipeline.py and that your pipeline printed ETL pipeline complete! at the end.

ⓧ I'd like to double check the full code

Your etl_pipeline.py file should contain these sections in order. Scroll through your file and confirm each is present:

  • Imports (pandas and psycopg2) and DB_CONFIG dictionary.
  • extract_staging_data() function.
  • transform_dim_date() function.
  • transform_dim_customer() and transform_dim_product() functions.
  • transform_fact_sales() function.
  • load_dimension() and load_fact() functions.
  • run_etl() function that calls extract, all four transforms, then all four loads in order.
  • if __name__ == "__main__": run_etl() at the bottom.
Run analytical queries against the star schema

Your warehouse is loaded. Now for the payoff. Remember the clunky queries from Step 3 that needed long GROUP BY clauses and repeated column references? Those same business questions now have clean, readable SQL thanks to the star schema's structure.

  • Create a new file called run_queries.py in your project folder.
  • Add the imports, config, and a reusable query runner function:
import pandas as pd
import psycopg2

DB_CONFIG = {
    "dbname": "retail_dw",
    "user": "postgres",
    "password": "postgres",
    "host": "localhost",
    "port": 5432,
}


def run_query(conn, title, query):
    # Print a formatted header, execute the query, display results
    print(f"\n{'=' * 50}")
    print(f"  {title}")
    print(f"{'=' * 50}")
    df = pd.read_sql(query, conn)
    print(df.to_string(index=False))
    print()
    return df

What does this code do?

The run_query helper takes a connection, a display title, and a SQL string. It executes the query, formats the result as a table, and prints it. This avoids repeating print logic for each analytical query.

  • Add the main function with the first two analytical queries below run_query:
def main():
    conn = psycopg2.connect(**DB_CONFIG)

    run_query(conn, "Monthly Revenue Summary", """
        SELECT
            d.year,
            d.month_name,
            d.month,
            COUNT(*) AS total_orders,
            SUM(f.quantity) AS units_sold,
            SUM(f.net_amount) AS revenue
        FROM warehouse.fact_sales f
        JOIN warehouse.dim_date d ON f.date_key = d.date_key
        GROUP BY d.year, d.month_name, d.month
        ORDER BY d.year, d.month
    """)

    run_query(conn, "Top 5 Customers by Revenue", """
        SELECT
            c.customer_name,
            c.customer_city,
            COUNT(*) AS total_orders,
            SUM(f.net_amount) AS total_revenue
        FROM warehouse.fact_sales f
        JOIN warehouse.dim_customer c ON f.customer_key = c.customer_key
        GROUP BY c.customer_name, c.customer_city
        ORDER BY total_revenue DESC
        LIMIT 5
    """)

What does this code do?

Each query joins the fact table to one dimension table using a single surrogate key. Compare this to the Step 3 queries where you had to GROUP BY every repeated column in the flat table.

The monthly revenue query joins fact_sales to dim_date to get calendar attributes. The top customers query joins to dim_customer for name and city.

  • Add the product category query directly below the Top 5 Customers query (still inside the main function):
    run_query(conn, "Revenue by Product Category", """
        SELECT
            p.product_category,
            p.product_subcategory,
            COUNT(*) AS total_orders,
            SUM(f.quantity) AS units_sold,
            SUM(f.net_amount) AS revenue,
            ROUND(AVG(f.discount_pct * 100), 1) AS avg_discount_pct
        FROM warehouse.fact_sales f
        JOIN warehouse.dim_product p ON f.product_key = p.product_key
        GROUP BY p.product_category, p.product_subcategory
        ORDER BY revenue DESC
    """)
  • Add the weekend vs weekday query and the script entry point below the category query:
    run_query(conn, "Weekend vs Weekday Sales", """
        SELECT
            CASE WHEN d.is_weekend THEN 'Weekend' ELSE 'Weekday' END AS day_type,
            COUNT(*) AS total_orders,
            SUM(f.net_amount) AS revenue,
            ROUND(AVG(f.net_amount), 2) AS avg_order_value
        FROM warehouse.fact_sales f
        JOIN warehouse.dim_date d ON f.date_key = d.date_key
        GROUP BY d.is_weekend
        ORDER BY d.is_weekend
    """)

    conn.close()
    print("All queries complete!")


if __name__ == "__main__":
    main()

What does this code do?

The category query joins to dim_product and groups by category and subcategory. The weekend vs weekday query uses dim_date.is_weekend to split sales into two groups with a single boolean.

Every query uses a simple JOIN on a surrogate key. No repeated columns. No verbose GROUP BY lists. This is the star schema paying off.

  • Save run_queries.py.

Before you run this, do you think the queries will return results or will something fail? Think about whether the warehouse tables actually have data from your ETL run.

  • Run the analytical queries:
python run_queries.py

You should see four formatted tables printed: Monthly Revenue Summary, Top 5 Customers by Revenue, Revenue by Product Category, and Weekend vs Weekday Sales. Each uses a clean JOIN to pull dimension attributes. The output ends with All queries complete!.

Seeing empty results or a table not found error?

If queries return empty DataFrames, the ETL pipeline did not load data. Re-run python etl_pipeline.py and check for errors in the Load phase output.

If you see a relation not found error, confirm the warehouse schema exists by running python create_star_schema.py before re-running the ETL pipeline. Help me debug my star schema queries

✔️ Awesome, I've got everything!

Great. Double check you have saved run_queries.py and that your output shows all four analytical tables followed by All queries complete! at the end.

ⓧ I'd like to double check the full code

Your run_queries.py file should contain these sections in order:

  • Imports (pandas and psycopg2) and DB_CONFIG dictionary.
  • run_query() helper function.
  • main() function with four run_query calls: Monthly Revenue Summary, Top 5 Customers by Revenue, Revenue by Product Category, Weekend vs Weekday Sales.
  • conn.close() and print statement at the end of main().
  • if __name__ == "__main__": main() at the bottom.

Your ETL pipeline is complete and your star schema is delivering clean, readable analytical results. Next up, you will take on the Secret Mission and add data quality validation to make this pipeline production-ready.

Secret mission

Add Data Quality Validation

Production ETL pipelines can receive dirty data with missing or invalid dimension keys. Can you implement referential integrity checks, data constraint validation, and the Kimball placeholder row technique to catch problems before they corrupt your warehouse?

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally with no cloud resources, so there are no ongoing costs.

Resources you used:

  • A local PostgreSQL database called retail_dw with staging and warehouse schemas.
  • Python project files: generate_data.py, setup_database.py, create_star_schema.py, etl_pipeline.py, run_queries.py, validate_data_quality.py, and raw_sales.csv.
  • A Python virtual environment with pandas and psycopg2-binary installed.
  • PostgreSQL 18 installed via the PGDG apt repository.

Keep everything running

No action needed. Choose this if you are still actively building or want to experiment with more queries against your star schema.

  • Your PostgreSQL service continues running in the background.
  • Your retail_dw database stays available for queries at any time.
  • Your Python virtual environment stays on disk, ready to reactivate.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your database files and project code stay on disk for later.

  • Deactivate your Python virtual environment by running:
deactivate
  • Stop the PostgreSQL service by running:
sudo systemctl stop postgresql

When you are ready to come back, restart PostgreSQL by running:

sudo systemctl start postgresql

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your database and project files.

  • Deactivate your virtual environment first by running:
deactivate
  • Drop the retail_dw database by running:
psql -U postgres -h localhost -c "DROP DATABASE retail_dw;"
  • Delete your project files and virtual environment by running:
rm generate_data.py setup_database.py create_star_schema.py etl_pipeline.py run_queries.py validate_data_quality.py raw_sales.csv
rm -rf venv
  • If you want to uninstall PostgreSQL 18 entirely, run:
sudo apt remove postgresql-18

Nice Work!

Nice Work!

You just built a production-style data warehouse from scratch. You took messy, flat retail data and transformed it into a clean Kimball star schema powered by your own Python ETL pipeline.

You've learned how to:

  • Build a complete ETL pipeline in Python that extracts raw data from a staging area, transforms it into proper dimensions and facts, and loads it into a star schema data warehouse.
  • Design a Kimball-style dimensional model with three dimension tables (date, customer, product) and one fact table (sales), connected by surrogate keys.
  • Experience firsthand why dimensional modeling exists by comparing clunky flat-data queries against clean, readable star schema queries using simple JOINs.
  • (Secret Mission) Implemented data quality validation with referential integrity checks and Kimball placeholder rows for handling missing dimension keys.

Ready to quiz yourself?