Build a Medallion Architecture Pipeline

Build a Bronze/Silver/Gold ETL pipeline using Python and DuckDB.

Introduction

30 Second Summary

Every time you buy something online, your order becomes a raw data point. Transforming those scattered records into clean, trustworthy dashboards is the job of a data pipeline.

In this project, you will build a Python-based ETL pipeline that ingests raw e-commerce CSV files and transforms them through three layers (Bronze, Silver, Gold) into business-ready analytics tables using DuckDB.

What You'll Build

You'll run a single command and watch messy order data flow through Bronze, Silver, and Gold layers, emerging as ranked customer reports and profit margin breakdowns you can query on the spot.

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

  • A pipeline you can re-run anytime that transforms raw CSV data into clean, queryable analytics tables in seconds.
  • Hands-on experience writing analytical SQL including JOINs, CTEs, and window functions that transfer directly to Microsoft Fabric and Azure.
  • A concrete understanding of the medallion architecture pattern used in every modern data lakehouse platform.
  • Secret Mission: Add a data quality validation layer that checks for anomalies between pipeline runs.

Do I need any prior experience?

This project starts from scratch. You don't need prior experience with Python, SQL, or data engineering. All setup steps are included in the guide.

Before We Start

Before We Start...

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

Set Up Your Environment

Before you can build a pipeline, you need a place to run it and data to flow through it.

In this step, you will install Python and DuckDB on your Windows machine, create your project folder, and populate it with the raw e-commerce CSV files your pipeline will transform.

In this step, get ready to:
  • Install Python 3.13 on Windows and verify it works.
  • Create the project folder structure and install DuckDB.
  • Create the raw e-commerce CSV data files.
Install Python 3.13 on Windows

Python is the language your pipeline script will be written in. DuckDB (installed next) runs as a Python package, so Python must be installed first.

  • Press the Windows key to open your search bar.
  • Type Command Prompt and press Enter to open it.
  • Check whether Python is already installed by running this command:
python --version

Look at the output and pick the tab that matches what you see.

✔️ I see Python 3.13

You should see something like Python 3.13.14 (the last number may differ).

  • Verify pip is available by running:
pip --version

You should see a version number and a path. Python is ready.

ⓧ I see a different Python version

If you see a version lower than 3.13 (such as Python 3.9.7), you need to install Python 3.13 alongside it.

  • Open your browser and go to https://www.python.org/downloads/.
  • Scroll down to the Python 3.13 section and click the Windows installer (64-bit) link to download it.
  • Run the downloaded installer.
  • On the very first screen, check the Add python.exe to PATH checkbox before clicking anything else.
  • Click Install Now.
  • Close the installer when it finishes.
  • Close your current Command Prompt window and open a new one (the PATH change only takes effect in new windows).
  • Verify the new version by running:
python --version

You should now see Python 3.13.14 or a similar 3.13 version.

  • Confirm pip is available by running:
pip --version

ⓧ Python is not recognized

This means Python is not installed on your machine yet.

  • Open your browser and go to https://www.python.org/downloads/.
  • Scroll down to the Python 3.13 section and click the Windows installer (64-bit) link to download it.
  • Run the downloaded installer.
  • On the very first screen, check the Add python.exe to PATH checkbox before clicking anything else.
  • Click Install Now.
  • Close the installer when it finishes.
  • Close your current Command Prompt window and open a new one (the PATH change only takes effect in new windows).
  • Verify Python works by running:
python --version

You should see Python 3.13.14 or a similar 3.13 version.

  • Confirm pip is available by running:
pip --version

Still seeing an older version or an error?

If python --version still shows an old version after installing 3.13, the old version's PATH entry is taking priority. Re-run the Python 3.13 installer, select Modify, and ensure the PATH option is checked.

If the Microsoft Store opens instead of showing a version, close it and follow the install steps above to install from python.org directly.

Help me fix my Python installation on Windows.

Create the project structure and install DuckDB

DuckDB is an analytical database engine that runs entirely inside your Python process. It handles the SQL queries that transform your data through each pipeline layer.

  • Move to your Desktop by running this command:
cd %USERPROFILE%\Desktop
  • Create your project folder and its data subfolder by running these commands:
mkdir medallion-pipeline
cd medallion-pipeline
mkdir data\raw
  • Confirm the folder structure exists by running:
dir

You should see a data folder listed in the output.

  • Install DuckDB by running:
pip install duckdb==1.5.4

What does this command do?

This installs DuckDB version 1.5.4 as a Python package. DuckDB gives you a full SQL engine that reads CSV files directly and supports window functions, CTEs, and schemas. These are all features you will use in your pipeline.

  • Verify DuckDB installed correctly by running:
python -c "import duckdb; print(duckdb.__version__)"

You should see 1.5.4 printed in the terminal.

Seeing an import error or version mismatch?

If you see ModuleNotFoundError: No module named 'duckdb', pip may have installed the package for a different Python version. Run pip --version and check that the path points to your Python 3.13 installation.

On some Windows machines, DuckDB requires the Microsoft Visual C++ Redistributable. If you see a DLL-related error, download it from Microsoft's official site and install it.

Help me troubleshoot my DuckDB installation.

Create the raw data files

Your pipeline needs raw data to ingest. You will create three CSV files representing a small e-commerce dataset: orders, customers, and products.

  • Open the medallion-pipeline folder in your code editor.
  • Create a new file at data\raw\orders.csv and paste the following content:
order_id,customer_id,product_id,order_date,quantity,unit_price,status
1001,C001,P001,2024-01-15,2,29.99,completed
1002,C002,P003,2024-01-16,1,49.99,completed
1003,C001,P002,2024-01-17,3,19.99,completed
1004,C003,P001,2024-01-18,1,29.99,pending
1005,C004,P004,2024-01-20,,79.99,completed
1006,C002,P005,2024-01-21,2,99.99,completed
1007,C005,P002,2024-01-22,1,19.99,cancelled
1008,C001,P003,2024-01-25,2,49.99,completed
1009,C003,P004,2024-02-01,1,79.99,completed
1010,C006,P001,2024-02-03,4,29.99,completed
1011,C004,P005,2024-02-05,1,99.99,completed
1012,C002,P002,2024-02-08,5,19.99,completed
1013,C005,P003,2024-02-10,2,49.99,completed
1014,C001,P004,2024-02-14,1,79.99,completed
1015,C003,P001,2024-02-15,3,29.99,completed

Why is some of this data messy?

Notice that row 1005 has an empty quantity field and row 1007 has a cancelled status. This is intentional. Real-world data is never perfectly clean. Your Silver layer will handle filtering these out later.

  • Create a new file at data\raw\customers.csv and paste the following content:
customer_id,name,email,city,signup_date
C001,Alice Johnson,alice@example.com,New York,2023-06-15
C002,Bob Smith,bob@example.com,Los Angeles,2023-08-22
C003,Carol Williams,carol@example.com,Chicago,2023-09-10
C004,David Brown,david@example.com,Houston,2023-11-01
C005,Eve Davis,eve@example.com,Phoenix,2024-01-05
C006,Frank Miller,frank@example.com,New York,2024-01-20
  • Create a new file at data\raw\products.csv and paste the following content:
product_id,product_name,category,cost_price
P001,Wireless Mouse,Electronics,12.50
P002,Notebook Set,Stationery,8.00
P003,USB-C Hub,Electronics,22.00
P004,Mechanical Keyboard,Electronics,35.00
P005,Standing Desk Mat,Office,45.00
  • Save all three files in your code editor.

Before you verify, predict what you expect to see when you list the contents of the data\raw folder. How many files should appear?

  • Back in your Command Prompt, confirm all three files exist by running:
dir data\raw

You should see three files listed: customers.csv, orders.csv, and products.csv.

Missing one or more files?

Make sure you saved the files inside the data\raw subfolder, not directly in medallion-pipeline. Check the file names end with .csv and not .csv.txt (some editors add a hidden .txt extension).

Help me find my missing CSV files.

Your environment is ready. You have Python, DuckDB, and raw data all in place. Next up, you will write the Bronze layer to ingest these CSV files into a database.

Build the Bronze Layer

Your project folder is ready with raw CSV data waiting to be processed. Now it is time to build the first layer of your pipeline.

In the medallion architecture, the Bronze layer is where raw data lands. Its job is to ingest source files exactly as they are, with no cleaning or transformation. This preserves a faithful copy of the original data so you can always trace back to the source.

In this step, get ready to:
  • Create pipeline.py with functions that connect to DuckDB and ingest CSV files.
  • Run the pipeline and verify that all three tables appear in the Bronze layer with the correct row counts.
Create pipeline.py and write the Bronze layer ingestion

The pipeline script needs two things: a way to connect to the database, and a function that reads each CSV file into a Bronze table.

  • In your code editor, create a new file called pipeline.py inside your medallion-pipeline folder.
  • Add the imports, database connection function, and the beginning of the Bronze layer function by pasting this code:
import duckdb
import os


def create_connection():
    """Create a persistent DuckDB database connection."""
    return duckdb.connect("warehouse.duckdb")


def build_bronze_layer(con):
    """Bronze Layer: Ingest raw CSV data as-is."""
    print("\n--- BRONZE LAYER ---")
    print("Ingesting raw data from CSV files...\n")

    con.execute("CREATE SCHEMA IF NOT EXISTS bronze")

    con.execute("DROP TABLE IF EXISTS bronze.orders")
    con.execute("DROP TABLE IF EXISTS bronze.customers")
    con.execute("DROP TABLE IF EXISTS bronze.products")

What does this code do?

  • duckdb.connect("warehouse.duckdb") creates a persistent database file on disk. Every table you create is saved inside this single file.
  • CREATE SCHEMA IF NOT EXISTS bronze creates a namespace to group all raw-data tables together.
  • The DROP TABLE IF EXISTS lines ensure the pipeline can re-run cleanly without "table already exists" errors.
  • Save pipeline.py.
  • Confirm the file was created by running this in your Command Prompt:
dir pipeline.py

You should see pipeline.py listed with today's date.

Don't see pipeline.py?

  • Make sure your Command Prompt is inside the medallion-pipeline folder. Run cd to check your current directory.
  • Confirm you saved the file in the correct location from your code editor (not inside the data subfolder).

Help me figure out why my pipeline.py file isn't showing up in the right folder.

  • In pipeline.py, add the CSV ingestion and row count verification below the DROP TABLE lines (still inside the build_bronze_layer function):
    con.execute("""
        CREATE TABLE bronze.orders AS
        SELECT * FROM read_csv('data/raw/orders.csv')
    """)

    con.execute("""
        CREATE TABLE bronze.customers AS
        SELECT * FROM read_csv('data/raw/customers.csv')
    """)

    con.execute("""
        CREATE TABLE bronze.products AS
        SELECT * FROM read_csv('data/raw/products.csv')
    """)

    orders_count = con.execute("SELECT COUNT(*) FROM bronze.orders").fetchone()[0]
    customers_count = con.execute("SELECT COUNT(*) FROM bronze.customers").fetchone()[0]
    products_count = con.execute("SELECT COUNT(*) FROM bronze.products").fetchone()[0]

    print(f"  bronze.orders:    {orders_count} rows")
    print(f"  bronze.customers: {customers_count} rows")
    print(f"  bronze.products:  {products_count} rows")

What does this code do?

  • read_csv('data/raw/orders.csv') is a built-in DuckDB function that reads a CSV file directly into a SQL query. No pandas or extra libraries needed.
  • CREATE TABLE ... AS SELECT * takes the CSV contents and stores them as a permanent table in the Bronze schema.
  • The SELECT COUNT(*) queries verify how many rows landed in each table, giving you immediate feedback that ingestion worked.

DuckDB uses forward slashes in file paths even on Windows, so 'data/raw/orders.csv' works correctly as-is.

  • Below the build_bronze_layer function (outside it, at the top level of the file), add the main function and entry point:
def main():
    """Run the complete medallion pipeline."""
    print("=" * 60)
    print("  MEDALLION ARCHITECTURE DATA PIPELINE")
    print("=" * 60)

    con = create_connection()

    build_bronze_layer(con)

    con.close()

    print("\n" + "=" * 60)
    print("  Pipeline complete! Database saved to warehouse.duckdb")
    print("=" * 60)


if __name__ == "__main__":
    main()

What does main() do?

  • main() is the entry point that orchestrates the pipeline. It opens the database connection, runs the Bronze layer build, then closes the connection.
  • if __name__ == "__main__" ensures main() only runs when you execute the file directly (not when importing it as a module).
  • Save pipeline.py.

✔️ Awesome, I've got everything!

Great. Confirm you have saved the file and you are ready to run it.

ⓧ I'd like to double check the full code

Scroll up and confirm your pipeline.py contains these sections in order:

  • Imports (import duckdb and import os).
  • The create_connection() function returning a connection to warehouse.duckdb.
  • The build_bronze_layer(con) function with schema creation, DROP TABLE statements, three CREATE TABLE statements using read_csv, and three row count prints.
  • The main() function calling create_connection() and build_bronze_layer(con).
  • The if __name__ == "__main__" block at the bottom.
Run the pipeline and inspect the Bronze layer

Your pipeline script is complete. Time to run it and see the Bronze layer come to life.

  • Before you run the pipeline, predict what you expect to see. How many rows should appear in each Bronze table?
  • Run the pipeline by executing this command:
python pipeline.py

You should see output like this:

============================================================
  MEDALLION ARCHITECTURE DATA PIPELINE
============================================================

--- BRONZE LAYER ---
Ingesting raw data from CSV files...

  bronze.orders:    15 rows
  bronze.customers: 6 rows
  bronze.products:  5 rows

============================================================
  Pipeline complete! Database saved to warehouse.duckdb
============================================================

Notice that bronze.orders has 15 rows. That includes the row with a missing quantity (order 1005) and the cancelled order (order 1007). The Bronze layer keeps everything, messy data included.

Why keep messy data in Bronze?

The Bronze layer is your safety net. By preserving raw data exactly as received, you can always re-process it if your cleaning logic changes later.

This is the same pattern used in Microsoft Fabric and Azure Databricks. Bronze tables are append-only archives of source data. Cleaning and business logic happen in the Silver and Gold layers, never at ingestion.

  • Confirm the database file was created by running:
dir warehouse.duckdb

You should see warehouse.duckdb listed. This single file holds your entire Bronze layer database.

Seeing an error when running the pipeline?

  • If you see ModuleNotFoundError: No module named 'duckdb', make sure you installed DuckDB in the previous step with pip install duckdb==1.5.4.
  • If you see a file not found error mentioning data/raw/orders.csv, confirm your Command Prompt is in the medallion-pipeline folder (not inside data or data\raw).
  • If you see an indentation error, check that the code inside build_bronze_layer is indented with 4 spaces consistently.

Help me debug my pipeline.py Bronze layer error.

Your Bronze layer is live. You have raw e-commerce data sitting in a real database, ready for transformation. Next up, you will clean this messy data and enrich it with JOINs in the Silver layer.

Build the Silver Layer

Your Bronze layer holds raw data exactly as it arrived from the source files. That includes messy records like orders with missing quantities and cancelled transactions.

The Silver layer is where data becomes trustworthy. You will clean out bad records, join related tables together, and compute new columns that make the data ready for analysis.

In this step, get ready to:
  • Write a Silver layer function that cleans and enriches order data using a CTE and JOINs.
  • Call the Silver layer from the main pipeline.
  • Run the pipeline and verify that messy rows were removed.
Add the Silver layer function to your pipeline

The Silver layer has one job: take the raw Bronze data and produce a single, clean, enriched table that downstream analytics can trust.

  • In pipeline.py, find the end of the build_bronze_layer function (the last print line with bronze.products).
  • Below that function, start a new build_silver_layer function by copying this code:
def build_silver_layer(con):
    """Silver Layer: Clean, validate, and enrich data."""
    print("\n--- SILVER LAYER ---")
    print("Cleaning and enriching data...\n")

    con.execute("CREATE SCHEMA IF NOT EXISTS silver")

    con.execute("DROP TABLE IF EXISTS silver.order_details")

What does this code do?

  • CREATE SCHEMA IF NOT EXISTS silver creates a separate namespace for your cleaned data, keeping it separate from the raw Bronze tables.
  • DROP TABLE IF EXISTS ensures you can re-run the pipeline without errors from duplicate tables.

Now for the core transformation. A CTE (Common Table Expression) lets you define a temporary result set, then query it in the same statement. You will use one to first clean the orders, then JOIN the cleaned rows with customer and product data.

  • Still inside build_silver_layer, add this SQL query directly below the DROP TABLE line:
    con.execute("""
        CREATE TABLE silver.order_details AS
        WITH cleaned_orders AS (
            SELECT
                order_id,
                customer_id,
                product_id,
                CAST(order_date AS DATE) AS order_date,
                quantity,
                unit_price,
                status
            FROM bronze.orders
            WHERE quantity IS NOT NULL
              AND status != 'cancelled'
        )
        SELECT
            co.order_id,
            co.order_date,
            co.quantity,
            co.unit_price,
            co.status,
            c.name AS customer_name,
            c.city AS customer_city,
            c.signup_date AS customer_signup_date,
            p.product_name,
            p.category,
            p.cost_price,
            (co.quantity * co.unit_price) AS total_amount,
            (co.quantity * co.unit_price) - (co.quantity * p.cost_price) AS profit
        FROM cleaned_orders co
        INNER JOIN bronze.customers c ON co.customer_id = c.customer_id
        INNER JOIN bronze.products p ON co.product_id = p.product_id
    """)

What does this SQL do?

This query does two things in a single statement using a CTE:

  • The WITH cleaned_orders AS block filters out rows where quantity is NULL and where status is cancelled. It also casts order_date to a proper DATE type.
  • The main SELECT uses two INNER JOIN clauses to combine orders with customers and products. Only rows that match across all three tables survive.
  • Two computed columns are added: total_amount (quantity times unit price) and profit (revenue minus cost). These columns did not exist in the raw data.

Why use a CTE instead of separate queries?

A CTE keeps the cleaning logic and the enrichment logic in one readable statement. The database engine optimises it as a single operation, so you do not pay the cost of writing intermediate results to disk.

In Microsoft Fabric and Azure Databricks, CTEs are the standard pattern for multi-step transformations within a single layer.

  • Still inside build_silver_layer, add these verification prints below the con.execute block:
    silver_count = con.execute("SELECT COUNT(*) FROM silver.order_details").fetchone()[0]
    bronze_count = con.execute("SELECT COUNT(*) FROM bronze.orders").fetchone()[0]

    print(f"  bronze.orders (raw):        {bronze_count} rows")
    print(f"  silver.order_details (clean): {silver_count} rows")
    print(f"  Removed {bronze_count - silver_count} rows (nulls + cancelled)")

What does this code do?

This compares the row count between Bronze and Silver. The difference tells you exactly how many rows were removed during cleaning. It is your first data quality signal: if the number looks wrong, you know the filters need adjusting.

  • Save pipeline.py.

Seeing a syntax error in the SQL?

  • Make sure the triple-quoted string starts with con.execute(""" and ends with """) on its own line. A missing quote or parenthesis will break the string.
  • Check that the indentation of the verification prints is at the same level as the con.execute lines (one level inside the function).
  • Help me fix a syntax error in my build_silver_layer function.
Call the Silver layer from main

The function exists but it will not run until you call it from main().

  • In pipeline.py, scroll down to the main() function.
  • Find this line:
    build_bronze_layer(con)

    con.close()
  • Add build_silver_layer(con) on a new line between build_bronze_layer(con) and con.close(), so it looks like this:
    build_bronze_layer(con)
    build_silver_layer(con)

    con.close()
  • Save pipeline.py.
Run and validate the Silver layer

Before you run the pipeline, think about what you expect to see. The Bronze layer has 15 rows in bronze.orders. One row has a null quantity (order 1005) and one is cancelled (order 1007). How many rows should the Silver layer contain?

  • Run the updated pipeline by executing this command:
python pipeline.py

You should see the Silver layer output showing 13 rows in silver.order_details and a message that 2 rows were removed (1 null quantity + 1 cancelled).

What is the Silver layer's role in a real pipeline?

The Silver layer is where data becomes trustworthy and queryable. Anyone who queries Silver can trust that nulls are handled, types are correct, and related tables are already joined.

In production data lakehouses like Microsoft Fabric, the Silver layer is the most-queried layer. Data analysts and BI tools connect here because they do not need to worry about data quality issues.

Seeing 0 rows or an error about missing tables?

  • If you see an error about bronze.orders not existing, make sure build_bronze_layer(con) is called BEFORE build_silver_layer(con) in your main() function.
  • If you see 0 rows, check that your CSV file paths in the Bronze layer use forward slashes: data/raw/orders.csv. DuckDB uses forward slashes even on Windows.
  • Help me debug my Silver layer pipeline output.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and that your output shows 13 Silver rows.

ⓧ I'd like to double check the full code

Scroll up through your pipeline.py and confirm these sections are present, in order:

  • import duckdb and import os at the top.
  • create_connection() function returning a connection to warehouse.duckdb.
  • build_bronze_layer(con) function that ingests 3 CSVs into the bronze schema.
  • build_silver_layer(con) function with the CTE, JOINs, computed columns, and row count prints.
  • main() calling build_bronze_layer(con) then build_silver_layer(con) before con.close().

Your data is now clean, joined, and enriched with computed columns. Next up, you will build the Gold layer where this trusted data gets transformed into business-ready analytics tables with rankings, aggregations, and growth metrics.

Build the Gold Layer

Your Silver layer has clean, enriched data that's ready for analysis. The next question is: what does this data actually tell us about the business?

The Gold layer answers specific business questions by aggregating and ranking the Silver data. In this step, you will build analytics tables using GROUP BY, window functions, and CTEs to produce dashboard-ready results.

In this step, get ready to:
  • Build Gold layer aggregation tables using GROUP BY and RANK().
  • Write an advanced analytics query using CTEs, ROW_NUMBER(), and LAG() window functions.
  • Run the full pipeline end-to-end and display formatted analytics results.
Build Gold layer aggregation tables

Gold layer tables answer specific business questions. Each table is pre-aggregated for a particular use case, like a daily sales report or a customer leaderboard.

  • In pipeline.py, find the end of the build_silver_layer function (after its last print statement).
  • Add the start of the Gold layer function below it by copying this code:
def build_gold_layer(con):
    """Gold Layer: Business-ready analytics tables."""
    print("\n--- GOLD LAYER ---")
    print("Building analytics tables...\n")

    con.execute("CREATE SCHEMA IF NOT EXISTS gold")

    con.execute("DROP TABLE IF EXISTS gold.daily_sales")
    con.execute("DROP TABLE IF EXISTS gold.customer_rankings")
    con.execute("DROP TABLE IF EXISTS gold.category_performance")
    con.execute("DROP TABLE IF EXISTS gold.customer_lifetime")

    # Daily sales summary
    con.execute("""
        CREATE TABLE gold.daily_sales AS
        SELECT
            order_date,
            COUNT(*) AS order_count,
            SUM(total_amount) AS daily_revenue,
            ROUND(AVG(total_amount), 2) AS avg_order_value
        FROM silver.order_details
        GROUP BY order_date
        ORDER BY order_date
    """)

What does this code do?

This creates a new gold schema and drops any existing Gold tables so the pipeline is idempotent (safe to re-run).

The gold.daily_sales table uses GROUP BY order_date to collapse individual orders into one row per day. COUNT(*), SUM(), and AVG() produce the daily metrics a dashboard would display.

  • Directly below that code, add the customer rankings and category performance tables:
    # Customer rankings by total spend
    con.execute("""
        CREATE TABLE gold.customer_rankings AS
        SELECT
            customer_name,
            customer_city,
            SUM(total_amount) AS total_spend,
            COUNT(*) AS order_count,
            ROUND(AVG(total_amount), 2) AS avg_order_value,
            RANK() OVER (ORDER BY SUM(total_amount) DESC) AS spend_rank
        FROM silver.order_details
        GROUP BY customer_name, customer_city
    """)

    # Category performance
    con.execute("""
        CREATE TABLE gold.category_performance AS
        SELECT
            category,
            SUM(total_amount) AS total_revenue,
            SUM(profit) AS total_profit,
            ROUND(SUM(profit) / SUM(total_amount) * 100, 1) AS profit_margin_pct,
            COUNT(*) AS total_orders
        FROM silver.order_details
        GROUP BY category
    """)

What are window functions?

RANK() OVER (ORDER BY SUM(total_amount) DESC) is a window function. It assigns a ranking number to each row based on total spend, without collapsing the rows into a single result like a plain GROUP BY would.

The gold.category_performance table calculates profit margin as a percentage. This is the kind of pre-computed metric a BI tool like Power BI would pull directly.

  • Now wire the Gold layer into the pipeline. In the main() function, find this line:
    build_silver_layer(con)

    con.close()
  • Add build_gold_layer(con) between build_silver_layer(con) and con.close(), so the section looks like this:
    build_silver_layer(con)
    build_gold_layer(con)

    con.close()
  • Save pipeline.py.
  • Run the pipeline to confirm the Gold layer builds by running this command:
python pipeline.py

You should see the Bronze and Silver output you saw before, followed by "--- GOLD LAYER ---" and "Building analytics tables..." printed to the console.

Seeing a NameError or syntax error?

  • Make sure build_gold_layer is defined as its own function at the module level (same indentation as build_silver_layer), not nested inside another function.
  • Check that the triple-quoted SQL strings (""") are closed properly. Each con.execute(""" needs a matching """).
  • Confirm build_gold_layer(con) appears in main() after build_silver_layer(con) and before con.close().

Help me debug my Gold layer function

Write an advanced analytics query using CTEs and window functions

The most powerful Gold layer table tracks how each customer's spending changes over time. This requires chaining two CTEs together and applying two different window functions.

  • In pipeline.py, add this code directly below the gold.category_performance query (still inside build_gold_layer):
    # Customer lifetime analysis with window functions
    con.execute("""
        CREATE TABLE gold.customer_lifetime AS
        WITH monthly_spend AS (
            SELECT
                customer_name,
                DATE_TRUNC('month', order_date) AS order_month,
                SUM(total_amount) AS monthly_total,
                COUNT(*) AS monthly_orders
            FROM silver.order_details
            GROUP BY customer_name, DATE_TRUNC('month', order_date)
        ),
        with_sequence AS (
            SELECT
                customer_name,
                order_month,
                monthly_total,
                monthly_orders,
                ROW_NUMBER() OVER (
                    PARTITION BY customer_name ORDER BY order_month
                ) AS month_number,
                LAG(monthly_total) OVER (
                    PARTITION BY customer_name ORDER BY order_month
                ) AS prev_month_total
            FROM monthly_spend
        )
  • Immediately below that (continuing the same con.execute call), add the final SELECT and the row-count print loop:
        SELECT
            customer_name,
            order_month,
            monthly_total,
            monthly_orders,
            month_number,
            prev_month_total,
            CASE
                WHEN prev_month_total IS NOT NULL AND prev_month_total > 0
                THEN ROUND((monthly_total - prev_month_total) / prev_month_total * 100, 1)
                ELSE NULL
            END AS mom_growth_pct
        FROM with_sequence
        ORDER BY customer_name, order_month
    """)

    print("  Gold layer tables created:")
    for table in ['daily_sales', 'customer_rankings', 'category_performance', 'customer_lifetime']:
        count = con.execute(f"SELECT COUNT(*) FROM gold.{table}").fetchone()[0]
        print(f"    gold.{table}: {count} rows")

What do these window functions do?

  • The first CTE (monthly_spend) groups orders by customer and month using DATE_TRUNC('month', order_date).
  • ROW_NUMBER() OVER (PARTITION BY customer_name ORDER BY order_month) assigns each customer's months a sequence number (1, 2, 3...) so you know which month was their first purchase.
  • LAG(monthly_total) OVER (...) grabs the previous row's value. This lets you compare this month's spend to last month's spend without a self-join.
  • The CASE expression calculates month-over-month growth as a percentage. It returns NULL for a customer's first month (no previous month to compare).
  • Save pipeline.py and run the pipeline again:
python pipeline.py

This time, under "--- GOLD LAYER ---" you should see row counts for all four tables. The print loop confirms each table was created successfully.

Seeing a SQL syntax error?

  • The two code chunks above form ONE continuous con.execute("""...""") call. Make sure there is no extra closing """) between them.
  • Check that the CASE block has a matching END keyword before AS mom_growth_pct.

Help me fix my customer_lifetime query

Run the full pipeline and display formatted results

The Gold tables exist, but numbers in a database are hard to read. A display function formats the results into clean tables so you can quickly interpret what the data tells you.

  • In pipeline.py, add this new function below build_gold_layer (above main()):
def display_sample_results(con):
    """Display sample queries from the Gold layer."""
    print("\n--- SAMPLE RESULTS ---\n")

    print("Top customers by spend:")
    print("-" * 60)
    results = con.execute("""
        SELECT spend_rank, customer_name, customer_city,
               ROUND(total_spend, 2) AS total_spend, order_count
        FROM gold.customer_rankings
        ORDER BY spend_rank
    """).fetchall()
    print(f"{'Rank':<6}{'Customer':<18}{'City':<15}{'Spend':>10}{'Orders':>8}")
    for row in results:
        print(f"{row[0]:<6}{row[1]:<18}{row[2]:<15}${row[3]:>8.2f}{row[4]:>8}")

What does this code do?

This function queries the Gold layer tables and prints formatted reports to the console. The f-string formatting (<6, >10) aligns columns into a readable table. This is what a BI dashboard would show, generated straight from SQL.

  • Continue the display_sample_results function by adding the category and customer growth reports directly below:
    print("\n\nCategory performance:")
    print("-" * 60)
    results = con.execute("""
        SELECT category, ROUND(total_revenue, 2) AS revenue,
               ROUND(total_profit, 2) AS profit, profit_margin_pct
        FROM gold.category_performance
        ORDER BY total_revenue DESC
    """).fetchall()
    print(f"{'Category':<15}{'Revenue':>12}{'Profit':>12}{'Margin':>10}")
    for row in results:
        print(f"{row[0]:<15}${row[1]:>10.2f}${row[2]:>10.2f}{row[3]:>9.1f}%")

    print("\n\nCustomer month-over-month growth:")
    print("-" * 60)
    results = con.execute("""
        SELECT customer_name, order_month, ROUND(monthly_total, 2) AS spend,
               mom_growth_pct
        FROM gold.customer_lifetime
        WHERE mom_growth_pct IS NOT NULL
        ORDER BY customer_name, order_month
    """).fetchall()
    print(f"{'Customer':<18}{'Month':<12}{'Spend':>10}{'MoM Growth':>12}")
    for row in results:
        print(f"{row[0]:<18}{str(row[1]):<12}${row[2]:>8.2f}{row[3]:>10.1f}%")

What does the MoM growth report show?

The WHERE mom_growth_pct IS NOT NULL filter removes each customer's first month (since there is no previous month to compare). The remaining rows show the percentage change in spending between consecutive months for each customer.

  • Now add display_sample_results(con) to main(). Find this line in your main() function:
    build_gold_layer(con)

    con.close()
  • Add display_sample_results(con) between build_gold_layer(con) and con.close(), so it looks like this:
    build_gold_layer(con)
    display_sample_results(con)

    con.close()
  • Save pipeline.py.

Before you run the full pipeline, what do you expect to see in the output? Think about how many layers will process, and what the final display tables will contain.

  • Run the complete pipeline end-to-end by running this command:
python pipeline.py

You should see the full output: Bronze ingestion, Silver cleaning, Gold table creation with row counts, and then three formatted reports showing customer rankings, category performance, and month-over-month growth percentages.

What should I see?

The "Top customers by spend" table shows Alice Johnson ranked #1 (she placed the most orders). The "Category performance" table shows Electronics with the highest revenue. The "Customer month-over-month growth" table shows percentage changes for customers who ordered in both January and February.

Seeing an error about a missing table?

  • If you see "Table with name customer_lifetime does not exist", the two chunks of the customer_lifetime query may not be joined correctly. They must form one continuous con.execute("""...""") call with no gap or extra quotes between them.
  • If you see "Table with name order_details does not exist", make sure build_silver_layer(con) still appears before build_gold_layer(con) in main().

Help me debug my pipeline output

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and your output shows all three formatted reports.

ⓧ I'd like to double check the full code

Your pipeline.py is now too long to display in a single block. Scroll up through your file and confirm these functions exist in this order:

  • create_connection() at the top.
  • build_bronze_layer(con) with three CSV ingestions.
  • build_silver_layer(con) with the CTE and JOINs.
  • build_gold_layer(con) with four CREATE TABLE statements (daily_sales, customer_rankings, category_performance, customer_lifetime) and a print loop.
  • display_sample_results(con) with three formatted reports (rankings, category, growth).
  • main() calling all four layer functions plus display_sample_results in order.

Your pipeline now transforms raw CSV data all the way through to formatted business analytics. This is the exact pattern used in Microsoft Fabric and Azure Databricks. Next up, you will tackle a Secret Mission that adds data quality validation between each layer.

Secret mission

Add a Data Quality Validation Layer

Your pipeline works, but how do you know the data flowing through it is trustworthy? In production, bad data silently corrupts downstream tables and dashboards. Add automated quality checks between each layer transition that catch anomalies before they propagate.

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, so there are no ongoing costs.

Resources you used:

  • medallion-pipeline project folder (contains pipeline.py, raw CSV data files, and warehouse.duckdb database).
  • DuckDB Python package (installed via pip).
  • Python 3.13 (installed in Step 1).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your pipeline or building on it.

  • The warehouse.duckdb file persists on disk. You can re-run pipeline.py or query the database anytime.
  • Python and DuckDB remain installed and available for other projects.

Pause - I'll come back to this later

Close any open terminals and your code editor. Your project files stay on disk and are ready whenever you return.

  • Close your Command Prompt window.
  • Close your code editor.
  • Your medallion-pipeline folder and warehouse.duckdb file remain on your Desktop. Open a new Command Prompt and run python pipeline.py from the project folder whenever you want to pick up again.

Delete - I don't want to use this again

Remove all project files and optionally uninstall the tools you set up.

  • Delete the project folder by opening a Command Prompt and running:
cd %USERPROFILE%\Desktop
rmdir /s /q medallion-pipeline
  • Uninstall the DuckDB package by running:
pip uninstall duckdb
  • Enter y when prompted to confirm.
  • Optionally uninstall Python 3.13. Open Settings, select Apps, then Installed apps.
  • Search for Python 3.13, click the three-dot menu next to it, and select Uninstall. Only do this if you have no other projects using Python.

Nice Work!

Nice Work!

Nice work! You just built a complete data pipeline that transforms raw e-commerce data through three layers of the medallion architecture using Python and DuckDB.

You've learned how to:

  • Built a complete ETL pipeline that ingests raw CSV data into a Bronze layer, cleans and enriches it with JOINs in a Silver layer, and produces business-ready analytics in a Gold layer.
  • Written analytical SQL using CTEs, window functions (RANK, ROW_NUMBER, LAG), and GROUP BY aggregations to answer real business questions.
  • Applied the exact medallion architecture pattern used in Microsoft Fabric, Azure Databricks, and Azure Synapse.
  • Secret Mission: Added a data quality validation layer that checks pipeline health between layer transitions.

Ready to quiz yourself?