Sales Data Pipeline with DuckDB

Build an ETL pipeline that cleans messy CSV data and loads it into DuckDB.

Introduction

30 Second Summary

Every company collects data, but raw data is messy. Dates arrive in different formats, numbers have dollar signs stuck to them, and entire fields go missing. Before anyone can make a business decision, someone has to clean that data up and put it somewhere useful.

In this project, you will build a complete data pipeline that takes a messy CSV file of sales records, cleans it with Python, and loads it into a DuckDB analytics database. You will then query that database with SQL to answer real business questions.

What You'll Build

A Python ETL script that extracts raw sales data from a CSV file, transforms it into clean records, and loads it into a DuckDB database for analytics.

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

  • A Python script that handles real-world data problems like missing values, inconsistent formats, and invalid records.
  • A persistent DuckDB database containing clean, structured sales data you can query at any time.
  • SQL analytics queries that calculate revenue by region, top products, and overall business performance.
  • Secret Mission: Add data quality monitoring that tracks how many records survive each stage of the pipeline.

Are there any prerequisites?

None. This project walks you through installing everything from scratch, including VS Code, Python, and DuckDB. No prior coding experience with Python or SQL is required.

Before We Start

To build a data pipeline, you need a code editor to write your Python script and a database library to store your cleaned data.

In this step, you will install VS Code, Python, and DuckDB so your machine is ready to build the pipeline.

In this step, get ready to:
  • Install VS Code and Python.
  • Create a project folder and virtual environment.
  • Install DuckDB.
Install VS Code and Python

VS Code is a free code editor where you will write and run your pipeline script. Python is the programming language that powers the pipeline.

Let's check if you already have VS Code installed.

  • Press the Windows key to open your search bar.
  • Type Visual Studio Code and look for a result.

✔️ I already have VS Code

Great. Open VS Code and move on to installing Python below.

ⓧ I don't see it in my search results

  • Open your browser and go to https://code.visualstudio.com/download.
  • Click the Windows button to download the User Setup installer.
  • Run the downloaded installer and follow the prompts to complete installation.
  • Once installed, press the Windows key, type Visual Studio Code, and press Enter to open it.

What is VS Code?

VS Code is a free code editor made by Microsoft. It has a built-in terminal, file explorer, and support for Python extensions.

Now let's check if Python is installed on your machine. Python is what executes your pipeline code.

  • In VS Code, open the terminal by clicking Terminal in the top menu bar.
  • Click New Terminal.
  • Check your Python version by running this command:
python --version

✔️ I see version 3.13 or higher

Python is ready. Move on to the next substep where you create your project folder.

ⓧ I see an older version

You need Python 3.13 or higher for this project. Download the latest version and install it alongside your existing version.

  • Open your browser and go to https://www.python.org/downloads/.
  • Click the large Download Python button at the top of the page.
  • Run the installer.
  • On the first screen, check the box that says Add python.exe to PATH.
  • Click Install Now and follow the prompts.
  • Close and reopen VS Code so the terminal picks up the new PATH.
  • Open a new terminal and run python --version again to confirm it shows 3.13 or higher.

Older version still showing?

If the old version still appears, your system PATH may be pointing to the previous installation. Try typing the full path to the new Python executable, or search Windows settings for Manage app execution aliases and ensure the python.exe alias points to the correct version.

ⓧ Command not found

Python is not installed yet. Let's install it now.

  • Open your browser and go to https://www.python.org/downloads/.
  • Click the large Download Python button at the top of the page.
  • Run the downloaded installer.
  • On the first screen, check the box that says Add python.exe to PATH. This is critical.
  • Click Install Now and follow the prompts until installation completes.
  • Close and reopen VS Code so the terminal picks up the new PATH.
  • Open a new terminal and run python --version to confirm it shows 3.13 or higher.

Why check the PATH box?

Adding Python to PATH means you can type python in any terminal and your system will find it. Without this, you would need to type the full file path to the Python executable every time.

Create a project folder and virtual environment

A virtual environment is an isolated space for your project's packages. It prevents conflicts between different projects on your machine.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir sales-pipeline
cd sales-pipeline
  • Confirm the folder was created by running:
ls

You should see an empty directory (no output yet, since the folder is new).

  • Open this folder in VS Code by running:
code .

What does code . do?

The code command launches VS Code from the terminal. The dot (.) tells it to open the current folder.

  • If a new VS Code window opened, use it from now on. Open a new terminal in that window by clicking Terminal then New Terminal.
  • Create a virtual environment by running this command:
python -m venv venv

What is a virtual environment?

The command python -m venv venv creates a folder called venv inside your project. This folder holds a private copy of Python and any packages you install.

This means packages you install here stay separate from any other Python projects on your machine.

  • Activate the virtual environment by running:
venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Seeing a script execution error?

On Windows, PowerShell may block the activation script by default. If you see an error mentioning that running scripts is disabled on this system, run this command first:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

Type Y and press Enter when prompted. Then run venv\Scripts\activate again.

Help me fix this PowerShell execution policy error.

Install DuckDB

DuckDB is an analytics database that runs inside your Python script. It can read CSV files directly with SQL and store results in a local database file.

  • Install DuckDB by running this command:
pip install duckdb==1.5.3

Why pin the version?

Using ==1.5.3 ensures everyone following this project gets the exact same version. This prevents unexpected differences in behavior from newer releases.

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

You should see 1.5.3 printed in the terminal.

Seeing an error about Visual C++?

DuckDB on Windows requires the Microsoft Visual C++ Redistributable. Most Windows 10 and 11 machines already have this installed.

If you see an error about missing DLL files, download and install the latest Visual C++ Redistributable from Microsoft's official site, then try the pip install command again.

Help me fix this DuckDB installation error on Windows.

Your development environment is ready. Next up, you will create your raw data source and write the first piece of the pipeline.

Set Up Your Tools

Before you can build a data pipeline, you need the right tools on your machine.

In this step, you will install VS Code (your code editor), Python (the programming language your pipeline runs on), and DuckDB (the analytics database that stores your clean data).

In this step, get ready to:
  • Install VS Code and Python on your Windows machine.
  • Create a project folder and set up an isolated Python environment.
  • Install the DuckDB package.
Install VS Code and Python

VS Code is the code editor where you will write your pipeline script. Python is the language that powers it.

  • Open your web browser and navigate to https://code.visualstudio.com/download.
  • Click the Windows button to download the User Setup installer.
  • Run the downloaded installer and follow the prompts to complete installation.
  • When the installer finishes, check the Launch Visual Studio Code option and click Finish.

What is VS Code?

VS Code is a free code editor from Microsoft. It has a built-in terminal, file explorer, and extensions that make writing Python much easier.

  • Next, open your browser and navigate to https://www.python.org/downloads/.
  • Click Download Python to download the latest Windows installer (64-bit).
  • Run the downloaded installer.
  • On the first screen of the installer, check the box that says Add python.exe to PATH. This is critical for running Python from the terminal.
  • Click Install Now and wait for the installation to complete.
  • Click Close when the installer finishes.

Why check Add python.exe to PATH?

PATH tells your terminal where to find programs. Without this checkbox, typing python in the terminal would fail because your computer would not know where Python is installed.

Now verify that Python is installed correctly.

  • Switch back to VS Code.
  • Open the built-in terminal by clicking Terminal in the top menu bar, then clicking New Terminal.
  • Check your Python version by running this command:
python --version

✔️ I see Python 3.13 or higher

You should see something like Python 3.13.x or Python 3.14.x. Either version works for this project. Python is ready to go.

ⓧ I see an older version

If you see a version below 3.13 (like 3.9 or 3.10), your system has an older Python that was installed previously.

  • Go back to https://www.python.org/downloads/ and download the latest installer.
  • Run the installer. Check Add python.exe to PATH and click Install Now.
  • Close and reopen the VS Code terminal, then run python --version again.

Still seeing the old version?

Open Start and search for Manage app execution aliases. Make sure the python.exe alias points to the Python version you just installed.

ⓧ Command not found

If you see an error that the command is not recognized, the PATH was not set during installation.

  • Re-run the Python installer you downloaded.
  • Select Modify.
  • On the next screen, make sure Add python.exe to PATH is checked.
  • Complete the installer, then close and reopen the VS Code terminal.
  • Run python --version again.
Create your project folder and virtual environment

A virtual environment is an isolated space where you install packages for one project without affecting anything else on your computer.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir sales-pipeline
cd sales-pipeline
  • Confirm the folder exists by running:
ls

You should see the sales-pipeline folder listed on your Desktop. Since you just moved into it, the terminal prompt now shows sales-pipeline as your current directory.

  • Now open this folder in VS Code by clicking File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the sales-pipeline folder, and click Select Folder.
  • If VS Code asks whether you trust the authors of the files in this folder, click Yes, I trust the authors.

VS Code reloads with sales-pipeline as the open workspace. Now open a fresh terminal.

  • Open the terminal by pressing Ctrl+` (the backtick key, above Tab).
  • Create a virtual environment by running this command:
python -m venv venv

What does this command do?

python -m venv venv tells Python to create a virtual environment in a folder called venv inside your project. Any packages you install will live here, separate from your system Python.

  • Activate the virtual environment by running:
venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Seeing a script execution error?

Windows PowerShell sometimes blocks script execution by default. If you see an error mentioning that scripts are disabled on this system, run this command first:

Set-ExecutionPolicy -Scope Process -ExecutionPolicy Bypass

This allows scripts to run for the current terminal session only. Then try the activate command again.

Why is my virtual environment not activating?

Install DuckDB

DuckDB is an analytics database designed for fast queries on structured data. Unlike traditional databases, it runs directly inside your Python script with no separate server to manage.

  • Install DuckDB into your virtual environment by running:
pip install duckdb==1.5.3

What does this command do?

pip install duckdb==1.5.3 tells pip (Python's package manager) to download and install version 1.5.3 of DuckDB. Pinning the version ensures your pipeline behaves consistently.

  • Verify the installation by running:
python -c "import duckdb; print(duckdb.__version__)"

You should see 1.5.3 printed in the terminal.

Not seeing 1.5.3?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If it is missing, run venv\Scripts\activate again before installing.

If you see an error about Microsoft Visual C++ Redistributable, download and install it from the official Microsoft website, then try the install command again.

Help me troubleshoot my DuckDB installation

Why DuckDB instead of SQLite?

SQLite is designed for transactional workloads (inserting one row at a time, looking up records by ID). DuckDB is built for analytics: scanning large datasets, grouping, and aggregating.

For a data pipeline that summarizes sales by region or calculates averages across thousands of rows, DuckDB is the right tool. It also reads CSV files directly with SQL, which simplifies extraction.

Your development environment is ready. Next up, you will create the raw data file that simulates messy real-world sales data coming into your pipeline.

Create Your Raw Data Source

Your tools are installed and your project folder is ready. Now it's time to get your hands on some data.

Real-world data is messy. It arrives with missing values, inconsistent formatting, and outright errors. In this step, you'll create a sample dataset that simulates these problems, then write the first stage of your pipeline to read it.

In this step, get ready to:
  • Create a raw CSV file with intentionally messy sales data.
  • Write the Extract function that reads the CSV into DuckDB.
  • Run the pipeline and confirm the raw data loads successfully.
Create the messy sales data

Every data pipeline starts with a source. In production, data arrives from databases, APIs, and file exports. Your source is a CSV file filled with the kinds of problems data engineers face daily.

  • In VS Code's Explorer sidebar, right-click inside the sales-pipeline folder and select New File.
  • Name the file sales_raw.csv and press Enter.
  • Paste the following data into the file:
date,product,quantity,revenue,region
2024-01-15,Widget A,10,$1234.56,North
01/20/2024,Widget B,5,$890.00,SOUTH
2024-02-10,Gadget C,3,$456.78,east
March 5 2024,Widget A,,-$200.00,North
2024-01-25,,8,$670.00,WEST
2024-03-01,Gadget C,2,$345.99,south
01/15/2024,Widget B,-3,$120.00,North
2024-02-28,Widget A,15,$2100.50,EAST
2024-03-10,Gadget C,7,$980.25,West
2024-01-30,Widget B,4,$560.00,north
2024-02-15,Widget A,0,$0.00,South
2024-03-20,Gadget C,6,$720.00,NORTH

What problems are hiding in this data?

This file has 12 rows of sales records, each with a different combination of data quality issues that your pipeline will need to handle.

  • Inconsistent date formats: 2024-01-15, 01/20/2024, and March 5 2024 all appear in the same column.
  • Missing values: some rows have empty product names or missing quantities.
  • Bad numbers: negative quantities (-3) and revenue with dollar signs ($1234.56).
  • Inconsistent casing: North, NORTH, north all represent the same region.
  • Save sales_raw.csv by pressing Ctrl+S.
  • Confirm the file appears in the Explorer sidebar under sales-pipeline.

Don't see the file in the sidebar?

Make sure you created the file inside the sales-pipeline folder and not a subfolder. Check that the file name is exactly sales_raw.csv with no extra spaces.

If the sidebar doesn't update, click the refresh icon at the top of the Explorer panel.

Help me find my CSV file in VS Code

Write the Extract function

ETL stands for Extract, Transform, Load. It's the pattern that powers every data pipeline. You'll start with the "E" stage, which reads raw data from its source.

The key insight: DuckDB can read CSV files directly with SQL using its read_csv function. No separate parsing library required.

  • In VS Code's Explorer sidebar, right-click inside the sales-pipeline folder and select New File.
  • Name the file pipeline.py and press Enter.
  • Add the imports and the extract function by pasting the following code:
import duckdb
from datetime import datetime


def extract(con):
    """Extract: Read raw CSV data into DuckDB."""
    print("=" * 50)
    print("STEP 1: EXTRACT")
    print("=" * 50)

    # Read the raw CSV, treating all columns as text to avoid auto-detection issues
    result = con.sql("""
        SELECT * FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """)

    # Count total rows extracted
    row_count = con.sql("""
        SELECT count(*) AS total_rows
        FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """).fetchone()[0]

    print(f"Extracted {row_count} rows from sales_raw.csv")
    print("\nSample of raw data:")
    # Display first 5 rows so we can see the messy data
    con.sql("""
        SELECT * FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
        LIMIT 5
    """).show()

    return row_count

What does this code do?

The extract function takes a DuckDB connection and uses it to read sales_raw.csv directly with SQL.

  • read_csv('sales_raw.csv', header = true, all_varchar = true) tells DuckDB to read the file, use the first row as column names, and treat every value as text.
  • The all_varchar = true parameter is critical. It prevents DuckDB from guessing data types on messy data, letting you handle type conversion yourself in the Transform step.
  • fetchone()[0] grabs the first column of the first row from the count query.
  • .show() prints the results in a formatted table directly to the terminal.
  • Now add the main function below the extract function by pasting this code at the bottom of pipeline.py:

def main():
    print("SALES DATA PIPELINE")
    print("=" * 50)
    print(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()

    # Create an in-memory DuckDB connection for staging work
    con = duckdb.connect()

    extracted = extract(con)

    con.close()


if __name__ == "__main__":
    main()

What does the main function do?

The main function is the entry point for your pipeline. It creates an in-memory DuckDB connection, calls extract, and then closes the connection.

duckdb.connect() without any arguments creates a temporary in-memory database. This is perfect for staging work because nothing is saved to disk yet.

  • Save pipeline.py by pressing Ctrl+S.

Seeing red underlines in VS Code?

If VS Code shows squiggly underlines under import duckdb, it may not be using the Python interpreter from your virtual environment.

Press Ctrl+Shift+P, type Python: Select Interpreter, and choose the one with venv in its path.

Help me fix Python import errors in VS Code

✔️ Awesome, I've got everything!

Great. Make sure you've saved pipeline.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py file at this point:

import duckdb
from datetime import datetime


def extract(con):
    """Extract: Read raw CSV data into DuckDB."""
    print("=" * 50)
    print("STEP 1: EXTRACT")
    print("=" * 50)

    result = con.sql("""
        SELECT * FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """)

    row_count = con.sql("""
        SELECT count(*) AS total_rows
        FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """).fetchone()[0]

    print(f"Extracted {row_count} rows from sales_raw.csv")
    print("\nSample of raw data:")
    con.sql("""
        SELECT * FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
        LIMIT 5
    """).show()

    return row_count


def main():
    print("SALES DATA PIPELINE")
    print("=" * 50)
    print(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()

    con = duckdb.connect()

    extracted = extract(con)

    con.close()


if __name__ == "__main__":
    main()
Run the pipeline

Time to see extraction in action. Running the script will read sales_raw.csv and display the messy data in your terminal.

  • Run the pipeline by entering this command in your terminal:
python pipeline.py
  • You should see output starting with SALES DATA PIPELINE followed by Extracted 12 rows from sales_raw.csv and a table showing the first 5 rows of raw data.

What should I see?

The table displays the first 5 rows exactly as they appear in your CSV. Notice the dollar signs in the revenue column, the inconsistent date formats, and the mixed casing in the region column.

These are the problems your Transform step will fix in the next step.

Seeing a FileNotFoundError?

This means DuckDB cannot find sales_raw.csv. Make sure your terminal's current directory is sales-pipeline (the same folder where sales_raw.csv lives).

Check your current directory by running cd with no arguments. If you're in the wrong folder, navigate back to your project by running:

cd ~/Desktop/sales-pipeline

Help me fix my FileNotFoundError when running the pipeline

Your extraction is working. You can see all the messy data flowing through. Next up, you'll write the Transform step to clean every one of those problems with SQL.

Clean and Transform the Data

Your pipeline can now extract data from the CSV file. But if you looked at the output, you probably noticed the data is a mess. Dates appear in three different formats, revenue values have dollar signs baked in, and region names are capitalized inconsistently.

This is the "T" in ETL. You will write SQL queries that clean, standardize, and validate every row before it moves further down the pipeline.

In this step, get ready to:
  • Write a transform function that normalizes dates, revenue, quantities, and region names.
  • Write a validate function that filters out invalid records.
  • Run the pipeline to see the cleaned data compared to the raw input.
Write the transform function

The transform function reads the same raw CSV, but this time it applies SQL logic to fix every column. DuckDB supports CASE expressions for conditional logic, REPLACE to strip unwanted characters, and TRY_CAST for safe type conversion that returns NULL instead of crashing on bad data.

  • In pipeline.py, find the end of your extract() function (the line return row_count).
  • Add the following transform() function below it, leaving a blank line between the two functions:
def transform(con):
    """Transform: Clean and standardize the raw data."""
    print("\n" + "=" * 50)
    print("STEP 2: TRANSFORM")
    print("=" * 50)

    con.sql("""
        CREATE OR REPLACE TABLE sales_staging AS
        SELECT
            CASE
                WHEN date LIKE '____-__-__' THEN TRY_CAST(date AS DATE)
                WHEN date LIKE '__/__/____' THEN TRY_CAST(
                    substr(date, 7, 4) || '-' || substr(date, 1, 2) || '-' || substr(date, 4, 2) AS DATE
                )
                ELSE NULL
            END AS sale_date,

            CASE
                WHEN trim(product) = '' OR product IS NULL THEN NULL
                ELSE trim(product)
            END AS product,

            CASE
                WHEN TRY_CAST(quantity AS INTEGER) < 0 THEN NULL
                WHEN TRY_CAST(quantity AS INTEGER) = 0 THEN NULL
                ELSE TRY_CAST(quantity AS INTEGER)
            END AS quantity,

            TRY_CAST(
                REPLACE(REPLACE(REPLACE(revenue, '$', ''), ',', ''), '-', '')
                AS DECIMAL(10, 2)
            ) AS revenue,

            CASE
                WHEN lower(trim(region)) = 'north' THEN 'North'
                WHEN lower(trim(region)) = 'south' THEN 'South'
                WHEN lower(trim(region)) = 'east' THEN 'East'
                WHEN lower(trim(region)) = 'west' THEN 'West'
                ELSE 'Unknown'
            END AS region

        FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """)

    total = con.sql("SELECT count(*) FROM sales_staging").fetchone()[0]
    print(f"Transformed {total} rows")

    print("\nSample of cleaned data:")
    con.sql("SELECT * FROM sales_staging LIMIT 5").show()

    return total

What does this code do?

  • CREATE OR REPLACE TABLE sales_staging AS SELECT ... creates a new in-memory table with the cleaned results of the query.
  • The first CASE block detects date formats using LIKE patterns. Dates matching YYYY-MM-DD get cast directly. Dates in MM/DD/YYYY format get rearranged with substr. Anything else (like "March 5 2024") becomes NULL.
  • TRY_CAST converts values to the target type (DATE, INTEGER, DECIMAL) but returns NULL instead of throwing an error when a value is invalid. This is essential for messy data.
  • Three nested REPLACE calls strip $, ,, and - from revenue before casting to a decimal number.
  • The region CASE block lowercases and trims each value, then maps it to a properly capitalized version (North, South, East, West).
  • Save pipeline.py.
  • You will not see a change yet. The next substep adds validation, and then you will wire everything together and run it.

Seeing red underlines in VS Code?

If VS Code highlights the SQL string with errors, that is just the editor struggling to parse SQL inside a Python string. It will still run correctly.

Double-check that your triple quotes """ are matched. A missing closing set is the most common cause of syntax errors in this block.

Help me fix a syntax error in my transform function's SQL string.

Add data validation

Even after cleaning, some records are too broken to be useful. A row with no product name or zero revenue does not belong in your analytics database. The validate function filters these out and reports how many survived.

  • Below your transform() function in pipeline.py, add the following validate() function:
def validate(con):
    """Validate: Filter out invalid records."""
    print("\n" + "=" * 50)
    print("STEP 3: VALIDATE")
    print("=" * 50)

    con.sql("""
        CREATE OR REPLACE TABLE sales_valid AS
        SELECT *
        FROM sales_staging
        WHERE product IS NOT NULL
          AND revenue IS NOT NULL
          AND revenue > 0
          AND sale_date IS NOT NULL
    """)

    valid_count = con.sql("SELECT count(*) FROM sales_valid").fetchone()[0]
    total_count = con.sql("SELECT count(*) FROM sales_staging").fetchone()[0]
    rejected_count = total_count - valid_count

    print(f"Valid records: {valid_count}")
    print(f"Rejected records: {rejected_count}")
    print(f"Pass rate: {valid_count / total_count * 100:.1f}%")

    return valid_count, rejected_count

What does this code do?

  • It creates a new table called sales_valid that only contains rows passing all four checks: product is not NULL, revenue is not NULL, revenue is greater than 0, and sale_date is not NULL.
  • It compares the row count of sales_valid against sales_staging to calculate how many records were rejected.
  • The pass rate percentage tells you at a glance how clean your source data is. In production, a sudden drop in pass rate is an early warning that something is wrong upstream.
  • Save pipeline.py.
  • You will not see a change yet. The next substep updates main() to call both new functions.

Indentation looking off?

Make sure def validate(con): starts at column 0 (no indentation). It is a top-level function, not nested inside transform().

Help me fix the indentation of my validate function.

Run the updated pipeline

Your two new functions exist but nothing calls them yet. You need to update main() to wire the full pipeline together: extract, transform, validate, then print a summary.

  • In pipeline.py, scroll down to your main() function and replace it entirely with this updated version:
def main():
    print("SALES DATA PIPELINE")
    print("=" * 50)
    print(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()

    # Create an in-memory DuckDB connection for staging work
    con = duckdb.connect()

    # Run each pipeline stage in order
    extracted = extract(con)
    transformed = transform(con)
    valid_count, rejected_count = validate(con)

    # Print a final summary of all stages
    print("\n" + "=" * 50)
    print("PIPELINE COMPLETE")
    print("=" * 50)
    print(f"Extracted: {extracted} rows")
    print(f"Transformed: {transformed} rows")
    print(f"Valid: {valid_count} rows")
    print(f"Rejected: {rejected_count} rows")

    con.close()


if __name__ == "__main__":
    main()

What changed in main?

  • transform(con) and validate(con) now run after extract(con), forming a three-stage pipeline.
  • Each function returns counts that feed into a final summary block. This gives you a quick dashboard of how many rows survived each stage.
  • Save pipeline.py.
  • Run the pipeline by entering this command in your terminal:
python pipeline.py

You should see three sections of output: the raw extraction, a sample of cleaned data (dates in a consistent format, revenue as plain numbers, regions properly capitalized), and a validation summary showing how many records passed.

Seeing a NameError or AttributeError?

If you see NameError: name 'transform' is not defined, make sure your transform() function is defined above main() in the file. Python reads top to bottom.

If you see a table-not-found error referencing sales_staging, confirm that transform(con) is called before validate(con) in main(). The validate function depends on the table that transform creates.

Help me debug my pipeline.py errors.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and your output shows all three stages completing.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py at this point. Compare your file against it:

import duckdb
from datetime import datetime


def extract(con):
    """Extract: Read raw CSV data into DuckDB."""
    print("=" * 50)
    print("STEP 1: EXTRACT")
    print("=" * 50)

    result = con.sql("""
        SELECT * FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """)

    row_count = con.sql("""
        SELECT count(*) AS total_rows
        FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """).fetchone()[0]

    print(f"Extracted {row_count} rows from sales_raw.csv")
    print("\nSample of raw data:")
    con.sql("""
        SELECT * FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
        LIMIT 5
    """).show()

    return row_count


def transform(con):
    """Transform: Clean and standardize the raw data."""
    print("\n" + "=" * 50)
    print("STEP 2: TRANSFORM")
    print("=" * 50)

    con.sql("""
        CREATE OR REPLACE TABLE sales_staging AS
        SELECT
            CASE
                WHEN date LIKE '____-__-__' THEN TRY_CAST(date AS DATE)
                WHEN date LIKE '__/__/____' THEN TRY_CAST(
                    substr(date, 7, 4) || '-' || substr(date, 1, 2) || '-' || substr(date, 4, 2) AS DATE
                )
                ELSE NULL
            END AS sale_date,

            CASE
                WHEN trim(product) = '' OR product IS NULL THEN NULL
                ELSE trim(product)
            END AS product,

            CASE
                WHEN TRY_CAST(quantity AS INTEGER) < 0 THEN NULL
                WHEN TRY_CAST(quantity AS INTEGER) = 0 THEN NULL
                ELSE TRY_CAST(quantity AS INTEGER)
            END AS quantity,

            TRY_CAST(
                REPLACE(REPLACE(REPLACE(revenue, '$', ''), ',', ''), '-', '')
                AS DECIMAL(10, 2)
            ) AS revenue,

            CASE
                WHEN lower(trim(region)) = 'north' THEN 'North'
                WHEN lower(trim(region)) = 'south' THEN 'South'
                WHEN lower(trim(region)) = 'east' THEN 'East'
                WHEN lower(trim(region)) = 'west' THEN 'West'
                ELSE 'Unknown'
            END AS region

        FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """)

    total = con.sql("SELECT count(*) FROM sales_staging").fetchone()[0]
    print(f"Transformed {total} rows")

    print("\nSample of cleaned data:")
    con.sql("SELECT * FROM sales_staging LIMIT 5").show()

    return total


def validate(con):
    """Validate: Filter out invalid records."""
    print("\n" + "=" * 50)
    print("STEP 3: VALIDATE")
    print("=" * 50)

    con.sql("""
        CREATE OR REPLACE TABLE sales_valid AS
        SELECT *
        FROM sales_staging
        WHERE product IS NOT NULL
          AND revenue IS NOT NULL
          AND revenue > 0
          AND sale_date IS NOT NULL
    """)

    valid_count = con.sql("SELECT count(*) FROM sales_valid").fetchone()[0]
    total_count = con.sql("SELECT count(*) FROM sales_staging").fetchone()[0]
    rejected_count = total_count - valid_count

    print(f"Valid records: {valid_count}")
    print(f"Rejected records: {rejected_count}")
    print(f"Pass rate: {valid_count / total_count * 100:.1f}%")

    return valid_count, rejected_count


def main():
    print("SALES DATA PIPELINE")
    print("=" * 50)
    print(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()

    con = duckdb.connect()

    extracted = extract(con)
    transformed = transform(con)
    valid_count, rejected_count = validate(con)

    print("\n" + "=" * 50)
    print("PIPELINE COMPLETE")
    print("=" * 50)
    print(f"Extracted: {extracted} rows")
    print(f"Transformed: {transformed} rows")
    print(f"Valid: {valid_count} rows")
    print(f"Rejected: {rejected_count} rows")

    con.close()


if __name__ == "__main__":
    main()

Your pipeline now extracts raw data, cleans it, and filters out the junk. Next up, you will load the surviving records into a persistent DuckDB database and run SQL analytics to answer real business questions.

Load into DuckDB and Query

Your pipeline can now extract, clean, and validate data. The cleaned records exist in memory, but they vanish the moment your script finishes running.

In this step, you will complete the pipeline by loading the validated data into a persistent DuckDB database file. Then you will write SQL queries that generate real business insights from your clean data.

In this step, get ready to:
  • Write a Load function that saves clean data to a persistent DuckDB database file.
  • Write SQL analytics queries that answer business questions about your sales data.
  • Run the full pipeline end-to-end and confirm data persists between runs.
Add the load function

The Load function is the "L" in ETL. It takes your validated in-memory data and writes it to a persistent database file on disk.

  • In pipeline.py, find the validate(con) function. Add the following load(con) function directly below it:
def load(con):
    """Load: Write clean data to persistent DuckDB database."""
    print("\n" + "=" * 50)
    print("STEP 4: LOAD")
    print("=" * 50)

    # Connect to a persistent database file (creates it if it doesn't exist)
    db = duckdb.connect("sales.duckdb")

    # Drop and recreate the table for a fresh load each run
    db.sql("DROP TABLE IF EXISTS sales_clean")
    db.sql("""
        CREATE TABLE sales_clean (
            sale_date DATE,
            product VARCHAR,
            quantity INTEGER,
            revenue DECIMAL(10, 2),
            region VARCHAR
        )
    """)

    # Fetch validated data from the in-memory connection
    clean_data = con.sql("SELECT * FROM sales_valid").fetchall()

    # Insert all rows into the persistent database
    db.executemany(
        "INSERT INTO sales_clean VALUES (?, ?, ?, ?, ?)",
        clean_data
    )

    loaded_count = db.sql("SELECT count(*) FROM sales_clean").fetchone()[0]
    print(f"Loaded {loaded_count} records into sales.duckdb")

    db.close()
    return loaded_count

What does this code do?

This function bridges the gap between in-memory processing and persistent storage.

  • duckdb.connect("sales.duckdb") opens (or creates) a database file on disk. Unlike the in-memory connection, data written here survives after the script ends.
  • DROP TABLE IF EXISTS ensures each run starts fresh, so you always load the latest validated data.
  • fetchall() pulls all rows from the in-memory sales_valid table as Python tuples.
  • executemany() inserts all those tuples into the persistent database in one batch.

Why two separate DuckDB connections?

The pipeline uses an in-memory connection (con) for staging and a file-based connection (db) for the final warehouse. This mirrors production pipelines where raw processing happens in temporary storage, and only clean data goes into the permanent database.

  • Save pipeline.py.

Seeing a syntax error?

Make sure the load(con) function is at the same indentation level as your other functions (extract, transform, validate). It should not be indented inside another function.

Help me debug a syntax error in my load function.

Write SQL analytics queries

With clean data loaded into a persistent database, you can now run SQL queries to answer business questions. This is the payoff of your entire pipeline.

  • In pipeline.py, add the following analyze() function below the load(con) function you just wrote:
def analyze():
    """Analyze: Run SQL queries on the loaded data."""
    print("\n" + "=" * 50)
    print("STEP 5: ANALYZE")
    print("=" * 50)

    # Open the persistent database for querying
    db = duckdb.connect("sales.duckdb")

    # Query 1: Revenue breakdown by region
    print("\n--- Revenue by Region ---")
    db.sql("""
        SELECT
            region,
            count(*) AS total_orders,
            SUM(revenue) AS total_revenue,
            ROUND(AVG(revenue), 2) AS avg_order_value
        FROM sales_clean
        GROUP BY region
        ORDER BY total_revenue DESC
    """).show()

    # Query 2: Top-selling products
    print("\n--- Top Products ---")
    db.sql("""
        SELECT
            product,
            count(*) AS total_orders,
            SUM(revenue) AS total_revenue
        FROM sales_clean
        GROUP BY product
        ORDER BY total_revenue DESC
    """).show()

    # Query 3: Overall pipeline summary statistics
    print("\n--- Overall Summary ---")
    db.sql("""
        SELECT
            count(*) AS total_orders,
            SUM(revenue) AS total_revenue,
            ROUND(AVG(revenue), 2) AS avg_revenue,
            MIN(sale_date) AS earliest_sale,
            MAX(sale_date) AS latest_sale
        FROM sales_clean
    """).show()

    db.close()

What do these queries do?

  • Revenue by Region groups orders by region and calculates total revenue, order count, and average order value for each.
  • Top Products shows which products generate the most revenue and orders.
  • Overall Summary gives a bird's-eye view of the dataset: total orders, total revenue, and the date range covered.
  • Save pipeline.py.

Getting a 'table sales_clean not found' error?

The analyze() function reads from the sales.duckdb file. If you run analyze() before load() has created that file, the table won't exist yet. Make sure your main() function calls load() before analyze().

Help me fix a 'table not found' error in my analyze function.

Update main and run the complete pipeline

Your new functions are ready, but main() doesn't call them yet. You need to wire up the Load and Analyze stages so the full pipeline runs end-to-end.

  • In pipeline.py, scroll down to the main() function.
  • Replace the entire main() function with this updated version:
def main():
    print("SALES DATA PIPELINE")
    print("=" * 50)
    print(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()

    # Create in-memory connection for staging work
    con = duckdb.connect()

    # Run all five pipeline stages in order
    extracted = extract(con)
    transformed = transform(con)
    valid_count, rejected_count = validate(con)
    loaded = load(con)
    analyze()

    # Print final pipeline summary
    print("\n" + "=" * 50)
    print("PIPELINE COMPLETE")
    print("=" * 50)
    print(f"Extracted: {extracted} rows")
    print(f"Transformed: {transformed} rows")
    print(f"Valid: {valid_count} rows")
    print(f"Rejected: {rejected_count} rows")
    print(f"Loaded: {loaded} rows")
    print(f"\nDatabase file: sales.duckdb")

    con.close()

What changed in main?

  • loaded = load(con) calls your new Load function and captures how many records were loaded.
  • analyze() runs the SQL queries against the persistent database. It takes no arguments because it opens its own connection to sales.duckdb.
  • The final summary now includes the loaded count and the database filename for confirmation.
  • Save pipeline.py.
  • Run the complete pipeline by running this command in your terminal:
python pipeline.py
  • You should see all five stages print their output: EXTRACT, TRANSFORM, VALIDATE, LOAD, and ANALYZE.
  • At the end, you should see a "Revenue by Region" table, a "Top Products" table, and an "Overall Summary" table printed in your terminal.
  • Confirm the database file was created by running:
ls
  • You should see sales.duckdb listed alongside pipeline.py and sales_raw.csv.
  • Run the pipeline a second time to confirm data persists between runs:
python pipeline.py
  • The output should be identical. The database is recreated each run with fresh data from the pipeline, confirming your Load stage works reliably.

Pipeline throws an error partway through?

Check that your sales_raw.csv file is still in the same folder as pipeline.py. DuckDB looks for it relative to where you run the command.

Also check that the load(con) and analyze() functions are placed between the validate(con) function and the main() function, not nested inside another function.

Help me debug my pipeline error.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py and that sales.duckdb appears in your project folder.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py file as it should look at the end of this step:

import duckdb
from datetime import datetime


def extract(con):
    """Extract: Read raw CSV data into DuckDB."""
    print("=" * 50)
    print("STEP 1: EXTRACT")
    print("=" * 50)

    result = con.sql("""
        SELECT * FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """)

    row_count = con.sql("""
        SELECT count(*) AS total_rows
        FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """).fetchone()[0]

    print(f"Extracted {row_count} rows from sales_raw.csv")
    print("\nSample of raw data:")
    con.sql("""
        SELECT * FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
        LIMIT 5
    """).show()

    return row_count


def transform(con):
    """Transform: Clean and standardize the raw data."""
    print("\n" + "=" * 50)
    print("STEP 2: TRANSFORM")
    print("=" * 50)

    con.sql("""
        CREATE OR REPLACE TABLE sales_staging AS
        SELECT
            CASE
                WHEN date LIKE '____-__-__' THEN TRY_CAST(date AS DATE)
                WHEN date LIKE '__/__/____' THEN TRY_CAST(
                    substr(date, 7, 4) || '-' || substr(date, 1, 2) || '-' || substr(date, 4, 2) AS DATE
                )
                ELSE NULL
            END AS sale_date,

            CASE
                WHEN trim(product) = '' OR product IS NULL THEN NULL
                ELSE trim(product)
            END AS product,

            CASE
                WHEN TRY_CAST(quantity AS INTEGER) < 0 THEN NULL
                WHEN TRY_CAST(quantity AS INTEGER) = 0 THEN NULL
                ELSE TRY_CAST(quantity AS INTEGER)
            END AS quantity,

            TRY_CAST(
                REPLACE(REPLACE(REPLACE(revenue, '$', ''), ',', ''), '-', '')
                AS DECIMAL(10, 2)
            ) AS revenue,

            CASE
                WHEN lower(trim(region)) = 'north' THEN 'North'
                WHEN lower(trim(region)) = 'south' THEN 'South'
                WHEN lower(trim(region)) = 'east' THEN 'East'
                WHEN lower(trim(region)) = 'west' THEN 'West'
                ELSE 'Unknown'
            END AS region

        FROM read_csv('sales_raw.csv', header = true, all_varchar = true)
    """)

    total = con.sql("SELECT count(*) FROM sales_staging").fetchone()[0]
    print(f"Transformed {total} rows")

    print("\nSample of cleaned data:")
    con.sql("SELECT * FROM sales_staging LIMIT 5").show()

    return total


def validate(con):
    """Validate: Filter out invalid records."""
    print("\n" + "=" * 50)
    print("STEP 3: VALIDATE")
    print("=" * 50)

    con.sql("""
        CREATE OR REPLACE TABLE sales_valid AS
        SELECT *
        FROM sales_staging
        WHERE product IS NOT NULL
          AND revenue IS NOT NULL
          AND revenue > 0
          AND sale_date IS NOT NULL
    """)

    valid_count = con.sql("SELECT count(*) FROM sales_valid").fetchone()[0]
    total_count = con.sql("SELECT count(*) FROM sales_staging").fetchone()[0]
    rejected_count = total_count - valid_count

    print(f"Valid records: {valid_count}")
    print(f"Rejected records: {rejected_count}")
    print(f"Pass rate: {valid_count / total_count * 100:.1f}%")

    return valid_count, rejected_count


def load(con):
    """Load: Write clean data to persistent DuckDB database."""
    print("\n" + "=" * 50)
    print("STEP 4: LOAD")
    print("=" * 50)

    db = duckdb.connect("sales.duckdb")

    db.sql("DROP TABLE IF EXISTS sales_clean")
    db.sql("""
        CREATE TABLE sales_clean (
            sale_date DATE,
            product VARCHAR,
            quantity INTEGER,
            revenue DECIMAL(10, 2),
            region VARCHAR
        )
    """)

    clean_data = con.sql("SELECT * FROM sales_valid").fetchall()

    db.executemany(
        "INSERT INTO sales_clean VALUES (?, ?, ?, ?, ?)",
        clean_data
    )

    loaded_count = db.sql("SELECT count(*) FROM sales_clean").fetchone()[0]
    print(f"Loaded {loaded_count} records into sales.duckdb")

    db.close()
    return loaded_count


def analyze():
    """Analyze: Run SQL queries on the loaded data."""
    print("\n" + "=" * 50)
    print("STEP 5: ANALYZE")
    print("=" * 50)

    db = duckdb.connect("sales.duckdb")

    print("\n--- Revenue by Region ---")
    db.sql("""
        SELECT
            region,
            count(*) AS total_orders,
            SUM(revenue) AS total_revenue,
            ROUND(AVG(revenue), 2) AS avg_order_value
        FROM sales_clean
        GROUP BY region
        ORDER BY total_revenue DESC
    """).show()

    print("\n--- Top Products ---")
    db.sql("""
        SELECT
            product,
            count(*) AS total_orders,
            SUM(revenue) AS total_revenue
        FROM sales_clean
        GROUP BY product
        ORDER BY total_revenue DESC
    """).show()

    print("\n--- Overall Summary ---")
    db.sql("""
        SELECT
            count(*) AS total_orders,
            SUM(revenue) AS total_revenue,
            ROUND(AVG(revenue), 2) AS avg_revenue,
            MIN(sale_date) AS earliest_sale,
            MAX(sale_date) AS latest_sale
        FROM sales_clean
    """).show()

    db.close()


def main():
    print("SALES DATA PIPELINE")
    print("=" * 50)
    print(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()

    con = duckdb.connect()

    extracted = extract(con)
    transformed = transform(con)
    valid_count, rejected_count = validate(con)
    loaded = load(con)
    analyze()

    print("\n" + "=" * 50)
    print("PIPELINE COMPLETE")
    print("=" * 50)
    print(f"Extracted: {extracted} rows")
    print(f"Transformed: {transformed} rows")
    print(f"Valid: {valid_count} rows")
    print(f"Rejected: {rejected_count} rows")
    print(f"Loaded: {loaded} rows")
    print(f"\nDatabase file: sales.duckdb")

    con.close()


if __name__ == "__main__":
    main()

Your pipeline is complete. You have built an end-to-end ETL system that extracts messy data, transforms it into a clean format, and loads it into a persistent analytics database. Next up, you will take on a secret mission to add data quality monitoring.

Secret mission

Add Data Quality Monitoring

Your pipeline works, but how do you know if data quality is getting better or worse over time? Production pipelines track execution metadata so teams can spot trends. Add observability to your pipeline by logging every run's stats to a dedicated table.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your local machine, so there are no ongoing costs or cloud bills to worry about. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • sales-pipeline project folder on your Desktop (contains pipeline.py, sales_raw.csv, sales.duckdb, and the venv virtual environment).
  • VS Code (installed application).
  • Python (installed application).

Keep everything running

No action needed. Choose this if you plan to keep building on this pipeline or want to reference the code later.

  • Your sales-pipeline folder stays on your Desktop with all files intact.
  • VS Code and Python remain installed and available for other projects.
  • The sales.duckdb database file is just a few KB and takes up negligible disk space.

Pause - I'll come back to this later

Deactivate the virtual environment to free up terminal resources. Your files stay on disk so you can resume later.

  • In your VS Code terminal, deactivate the virtual environment by running:
deactivate
  • Close VS Code.
  • When you come back, open the sales-pipeline folder in VS Code, open a terminal, and reactivate with venv\Scripts\activate.

Delete - I don't want to use this again

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

Delete the project folder
  • Close VS Code if it is still open.
  • Open a terminal (press the Windows key, type cmd, and press Enter).
  • Delete the project folder by running:
rmdir /s /q %USERPROFILE%\Desktop\sales-pipeline
  • This removes pipeline.py, sales_raw.csv, sales.duckdb, and the venv virtual environment (including the DuckDB pip package installed inside it).
Uninstall VS Code (optional)

Skip this if you plan to use VS Code for other projects.

  • Open Settings (press Windows + I).
  • Select Apps, then Apps & features.
  • Find Visual Studio Code in the list and select Uninstall.
  • Follow the prompts to complete the uninstall.
Uninstall Python (optional)

Skip this if you plan to use Python for other projects.

  • Open Settings (press Windows + I).
  • Select Apps, then Apps & features.
  • Find Python in the list and select Uninstall.
  • Follow the prompts to remove Python.

Nice Work!

Nice Work!

You just built a complete data pipeline from scratch using Python and DuckDB. From messy, inconsistent CSV data to clean, queryable analytics in a persistent database.

You've learned how to:

  • Build an ETL pipeline that extracts messy CSV data, transforms it with SQL (CASE expressions, REPLACE, and TRY_CAST), validates records, and loads clean data into a persistent DuckDB analytics database.
  • Handle real-world data quality issues including inconsistent date formats, missing values, negative quantities, and messy string formatting.
  • Query your data warehouse with SQL to generate business insights like revenue by region, top products, and overall sales summaries.
  • Secret Mission: Added pipeline observability with execution logging and data quality metrics tracked in a pipeline_runs table.

Ready to quiz yourself?