Introduction
Building your first data pipeline is one of those things that sounds more complicated than it is - until you're actually doing it, and then it's slightly more complicated than you expected. That's not a bad thing. The friction you hit along the way is where the real learning happens.
This tutorial walks through building a complete ETL pipeline in Python: extracting sales data from a CSV file, cleaning and validating it, and loading it into a PostgreSQL database. I've kept it realistic rather than toy-simple. The patterns here - the way the steps are structured, how validation is handled, how the orchestration is wired up - are the same patterns you'd use in production.
By the end you'll have a working pipeline and a much clearer mental model of how the pieces fit together.
What We're Building
A pipeline that runs daily and:
- Extracts sales data from CSV files dropped into a folder
- Cleans and validates the data
- Calculates key metrics
- Loads the results into PostgreSQL
- Runs on a schedule with automatic retries
Not glamorous, but this exact shape of pipeline exists in almost every company that deals with data. Get comfortable with it and you'll recognise it everywhere.
Prerequisites
- Python 3.8+
- PostgreSQL installed locally (or a free cloud instance)
- Basic familiarity with Pandas and SQL
If you're not there yet, Getting Started with Python Data Engineering covers the foundational skills worth having before diving in here.
A Note on Pipeline Design
Before writing any code, it's worth understanding what a good pipeline actually looks like. Three properties matter most:
Idempotent - running the pipeline twice should produce the same result as running it once. No duplicates, no inconsistent state. This sounds obvious but is easy to get wrong.
Observable - when something goes wrong (and it will), you need to know what happened. Good logging and meaningful error messages are not optional.
Fail loudly - a pipeline that silently loads bad data is worse than one that crashes. Validate early, fail fast, and make the error message tell you exactly what went wrong.
Keep these in mind as we go through the steps.
Step 1: Extract Data
The extract step is deceptively simple. You're just reading a file - but a few decisions here will save you pain later.
import pandas as pd from pathlib import Path def extract_sales_data(file_path: str) -> pd.DataFrame: df = pd.read_csv(file_path, dtype=str) # read everything as string first print(f"Extracted {len(df)} records from {file_path}") return df
Notice we're reading everything as strings with dtype=str. This is a habit worth building. If you let Pandas infer types on load, it will silently coerce things - a column with mostly integers but one stray "N/A" string will become a float column with a NaN in it. Reading as strings first and coercing types explicitly in the transform step gives you control over what happens.
Also worth noting: pathlib.Path over raw string paths. It handles OS differences cleanly and makes path manipulation much more readable.
Step 2: Transform Data
This is where most of the logic lives. The transform step is responsible for taking raw, potentially messy data and producing something clean and consistent.
def transform_sales_data(df: pd.DataFrame) -> pd.DataFrame: # coerce types explicitly now that we know what we're working with df['date'] = pd.to_datetime(df['date'], errors='coerce') df['quantity'] = pd.to_numeric(df['quantity'], errors='coerce') df['price'] = pd.to_numeric(df['price'], errors='coerce') # drop duplicates before anything else df = df.drop_duplicates(subset=['order_id']) # calculate derived fields df['revenue'] = df['quantity'] * df['price'] df['month'] = df['date'].dt.to_period('M') return df
The errors='coerce' argument on pd.to_datetime and pd.to_numeric is important. Instead of crashing on unparseable values, it replaces them with NaN. That gives your validation step something to catch rather than letting the error bubble up in a confusing way.
Drop duplicates early, before you calculate anything. Downstream aggregations on duplicate rows produce wrong numbers, and catching the problem at the source is always cleaner than debugging it later.
Step 3: Validate Data
This step gets skipped more often than it should, usually because it feels like extra work when you're in a hurry. Don't skip it. A pipeline that loads garbage data quietly is genuinely worse than one that fails loudly.
from pydantic import BaseModel, field_validator from typing import Optional import pandas as pd class SalesRecord(BaseModel): order_id: str date: str product: str quantity: int price: float @field_validator('quantity') @classmethod def quantity_must_be_positive(cls, v): if v <= 0: raise ValueError(f'quantity must be positive, got {v}') return v @field_validator('price') @classmethod def price_must_be_positive(cls, v): if v <= 0: raise ValueError(f'price must be positive, got {v}') return v def validate_data(df: pd.DataFrame) -> pd.DataFrame: # first drop rows where coercion failed (NaN in required columns) required = ['order_id', 'date', 'quantity', 'price', 'product'] before = len(df) df = df.dropna(subset=required) dropped = before - len(df) if dropped > 0: print(f"Warning: dropped {dropped} rows with missing required fields") # validate remaining rows with Pydantic valid_rows = [] for record in df.to_dict('records'): try: SalesRecord(**{k: record[k] for k in required}) valid_rows.append(record) except Exception as e: print(f"Invalid record {record.get('order_id', '?')}: {e}") return pd.DataFrame(valid_rows)
A couple of things worth calling out here. Rather than failing the entire pipeline on the first bad record, this version logs invalid rows and continues with the clean ones. Whether that's the right behaviour depends on your context - sometimes you want to fail hard on any bad data, sometimes you want to load what you can and report on the rest. Make that a conscious decision, not an accident.
For a deeper look at validation patterns, Financial Transaction Validation with Pydantic is worth working through.
Step 4: Load to Database
The load step has one job: get the clean data into the database reliably. A few things to get right here.
from sqlalchemy import create_engine, text import pandas as pd def load_to_postgres(df: pd.DataFrame, table_name: str) -> None: engine = create_engine( 'postgresql://user:password@localhost:5432/sales_db' ) with engine.begin() as conn: # delete today's data before inserting - makes the step idempotent conn.execute( text(f"DELETE FROM {table_name} WHERE month = :month"), {"month": str(df['month'].iloc[0])} ) df.to_sql(table_name, conn, if_exists='append', index=False) print(f"Loaded {len(df)} records into {table_name}")
The delete-before-insert pattern is how you make the load step idempotent. If the pipeline runs twice for the same day, the second run deletes the first run's data before inserting again. No duplicates.
We're also using engine.begin() which wraps everything in a transaction. If the insert fails halfway through, the delete is rolled back too. You never end up with a half-loaded table.
For more on working with databases in Python, Database Operations with SQLAlchemy walks through the patterns in detail.
Step 5: Orchestrate with Airflow
So far we have four functions that each do one thing. Now we wire them together and put them on a schedule with Apache Airflow.
from airflow.decorators import dag, task from datetime import datetime, timedelta @dag( schedule='@daily', start_date=datetime(2025, 1, 1), catchup=False, default_args={ 'retries': 2, 'retry_delay': timedelta(minutes=5), } ) def sales_etl_pipeline(): @task() def extract(): return extract_sales_data('/data/sales/latest.csv') @task() def transform(raw_df): return transform_sales_data(raw_df) @task() def validate(clean_df): return validate_data(clean_df) @task() def load(valid_df): load_to_postgres(valid_df, 'sales_daily') load(validate(transform(extract()))) sales_etl_pipeline()
The retries=2 and retry_delay in default_args mean that if any task fails, Airflow will retry it twice with a 5-minute gap. This handles transient issues - a brief network blip or a slow database - without waking you up at 3am.
catchup=False is worth understanding. If you deploy this DAG today but set start_date to a month ago, Airflow would by default try to backfill every daily run since then. catchup=False tells it to start from now instead.
If Airflow feels heavy for what you're building, Pokemon ETL Pipeline with Prefect shows the same pipeline pattern with Prefect, which has a much lighter local setup.
Putting It All Together
Here's the complete pipeline as a standalone script you can run without Airflow, useful for testing:
from pathlib import Path def run_pipeline(file_path: str, table_name: str = 'sales_daily') -> None: print("Starting pipeline...") raw = extract_sales_data(file_path) transformed = transform_sales_data(raw) validated = validate_data(transformed) if len(validated) == 0: raise ValueError("No valid records to load - aborting") load_to_postgres(validated, table_name) print(f"Pipeline complete. Loaded {len(validated)} records.") if __name__ == "__main__": run_pipeline("/data/sales/latest.csv")
The empty-dataframe check before loading is easy to overlook but important. If validation filters out everything, you don't want to run a delete against the database and then load nothing.
Common Mistakes to Avoid
A few things I've seen trip people up repeatedly when building pipelines like this:
Not thinking about idempotency from the start. The most common version of this is using if_exists='replace' in to_sql without realising it drops and recreates the table. That wipes your indexes, constraints, and any other configuration. Use if_exists='append' and handle deduplication explicitly.
Swallowing exceptions silently. A bare except: pass anywhere in a pipeline is a landmine. You'll come back three weeks later wondering why a table has stale data and have no logs to help you.
Hardcoding credentials. The database URL in the load step should come from an environment variable, not be written directly in the code. Use os.environ.get('DATABASE_URL') and store secrets somewhere safe.
Not testing with bad data. Your CSV will eventually arrive with a missing column, a column in the wrong order, or encoding issues. Test your extract and validate steps against intentionally broken inputs before you deploy.
Next Steps
Once this pipeline is running, a few natural directions to take it:
- Incremental loading - instead of processing the full CSV every day, track a watermark and only load new records. Essential once your data volumes grow.
- Data quality checks - add monitoring with Great Expectations to alert you when data distributions shift unexpectedly.
- Scale up - when your data outgrows single-machine Pandas, E-commerce Data Processing with PySpark shows the same ETL pattern at scale.
- Simpler ingestion - if you're pulling from APIs rather than files, Weather Data Pipeline with DLT shows how DLT handles schema inference and incremental loading automatically.
FAQ
Do I need PostgreSQL specifically, or can I use something else?
The patterns here work with any relational database. SQLite is fine for local development and learning - just change the connection string. For production, PostgreSQL, MySQL, and cloud warehouses like BigQuery or Snowflake all work with SQLAlchemy.
Should I use Airflow for every pipeline?
No. Airflow has real setup overhead. For a single pipeline running on a schedule, a cron job or a simple cloud scheduler is often enough. Airflow starts making sense when you have multiple pipelines with dependencies between them, and you need visibility into what's running and what failed.
What's the difference between ETL and ELT?
ETL (Extract, Transform, Load) transforms data before loading it into the destination - which is what this tutorial does. ELT (Extract, Load, Transform) loads raw data first and transforms it inside the warehouse using SQL or dbt. ELT has become more common as cloud warehouses got cheap enough that storing raw data is no longer a concern. Both approaches are valid and the right choice depends on your setup.
How do I handle pipeline failures in production?
At minimum: structured logging, alerting on task failures (Airflow and Prefect both support this natively), and making sure every step is idempotent so you can safely re-run after fixing a bug. As things get more complex, a dead-letter queue for bad records and proper monitoring dashboards become worth the investment.
My pipeline works locally but fails on the server. Where do I start?
Nine times out of ten it's one of: environment variables not set, a Python package version mismatch, or a file path that works on your machine but doesn't exist on the server. Check those three first before anything else.