Build Your First ETL Pipeline with Python and SQL (2026)

The best way to understand ETL is to build one. In this tutorial you will create a complete, working ETL pipeline using Python and SQL — from raw CSV data to a queryable SQLite database. Every step maps directly to the patterns used in production pipelines.

Prerequisites: Basic Python knowledge. No prior data engineering experience needed.

Setup

bash

pip install pandas sqlalchemy

Project structure:

text

etl_project/
├── etl.py
└── data/
    └── sales_data.csv

Step 1: Extract

Extraction pulls raw data from the source. Here the source is a CSV file — in production this would be a database, API, or cloud storage bucket. The pattern is identical.

Python

import pandas as pd

def extract(file_path: str) -> pd.DataFrame:
    df = pd.read_csv(file_path)
    print(f"[EXTRACT] {len(df)} records loaded from {file_path}")
    return df

Step 2: Transform

This is the most important stage — and where most ETL bugs hide. Always validate record counts before and after — unexpected drops show you exactly where data is being lost.

Python

def transform(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()

    # Drop rows missing critical fields
    df.dropna(subset=['order_id', 'customer_id'], inplace=True)

    # Fix data types — coerce bad values to NaN, then fill
    df['quantity'] = pd.to_numeric(df['quantity'], errors='coerce').fillna(0).astype(int)
    df['price']    = pd.to_numeric(df['price'],    errors='coerce').fillna(0.0)

    # Standardize dates
    df['order_date'] = pd.to_datetime(df['order_date'], errors='coerce')
    df.dropna(subset=['order_date'], inplace=True)

    # Remove duplicates
    df.drop_duplicates(subset=['order_id'], inplace=True)

    # Calculate derived field
    df['total_revenue'] = df['quantity'] * df['price']

    print(f"[TRANSFORM] {len(df)} clean records")
    return df

Step 3: Load

Use if_exists='replace' for full loads. Use if_exists='append' for incremental loads that add records to an existing table.

Python

from sqlalchemy import create_engine

def load(df: pd.DataFrame, table_name: str, db_path: str = "warehouse.db"):
    engine = create_engine(f"sqlite:///{db_path}")
    df.to_sql(
        name=table_name,
        con=engine,
        if_exists='replace',
        index=False
    )
    print(f"[LOAD] {len(df)} records → '{table_name}' in {db_path}")

Step 4: Orchestrate

Python

def run_etl():
    raw_df   = extract("data/sales_data.csv")
    clean_df = transform(raw_df)
    load(clean_df, table_name="fact_sales")

if __name__ == "__main__":
    run_etl()

bash

python etl.py

# [EXTRACT] 5000 records loaded from data/sales_data.csv
# [TRANSFORM] 4837 clean records
# [LOAD] 4837 records → 'fact_sales' in warehouse.db

Step 5: Query the Warehouse

SQL

-- Total revenue by product
SELECT product, SUM(total_revenue) AS revenue
FROM fact_sales
GROUP BY product
ORDER BY revenue DESC;

-- Top 10 customers
SELECT customer_id, SUM(total_revenue) AS total_spend
FROM fact_sales
GROUP BY customer_id
ORDER BY total_spend DESC
LIMIT 10;

Next Steps

Swap CSV for a real database using pd.read_sql() + SQLAlchemy connection string
Add logging with Python's logging module instead of print statements
Schedule it with cron or Apache Airflow
Add tests using pytest to validate data at each stage — covered in the Python ETL Testing Workflow post
Plan your career path — see what to learn next in our ETL developer course guide

Learn to test what you build

Building a pipeline is step one. The ETL Testing Course teaches you how to validate every stage with SQL and Python — ensuring your pipeline delivers trustworthy data every time it runs.

Frequently Asked Questions

What Python libraries do I need to build an ETL pipeline?

The core libraries are pandas (for data extraction and transformation) and SQLAlchemy (for loading data into a database). Install them with: pip install pandas sqlalchemy. For production pipelines you might also use psycopg2 for PostgreSQL connections.

What is the difference between if_exists replace and append in pandas to_sql?

if_exists='replace' drops the existing table and recreates it — used for full loads. if_exists='append' adds new rows without deleting old ones — used for incremental loads.

How do I schedule a Python ETL pipeline?

For simple scheduling, use cron on Linux/Mac or Windows Task Scheduler. For production workflows with monitoring and retry logic, use Apache Airflow, Prefect, or AWS Lambda with EventBridge.

Written by

Asim Noaman Lodhi

Certified Google Partner · QA Consultant · 12+ Years IT

QA consultant specializing in ETL testing and data quality. Trained 913+ students to transition into data testing roles through hands-on, real-world instruction.

4.5 Rating 913+ Students Google Partner 79 Lectures

Build Your First ETL Pipeline (Python + SQL)

Setup

Step 1: Extract

Step 2: Transform

Step 3: Load

Step 4: Orchestrate

Step 5: Query the Warehouse

Next Steps

Frequently Asked Questions

Asim Noaman Lodhi

Test Your ETL Pipelines Like a Pro