Building ETL Pipelines with Python – Airflow, dbt & Great Expectations 2026

Data pipelines are the plumbing of data science. No matter how good your model is, it is useless without reliable, clean, timely data flowing into it. This guide builds production-grade ETL (Extract, Transform, Load) pipelines using Python — with Apache Airflow for orchestration, dbt for SQL transformations, and Great Expectations for automated data quality checks.

Pipeline Architecture Patterns

ETL (Extract-Transform-Load) extracts raw data, transforms it in memory, then loads it to the destination — good when transformations are complex Python logic. ELT (Extract-Load-Transform) loads raw data first, then transforms it inside the warehouse using SQL — modern data warehouses (Snowflake, BigQuery, Redshift) are optimised for this pattern. Most modern data stacks use ELT with dbt for transformations and Python for extraction and orchestration.

Simple ETL Pipeline

import pandas as pd
import numpy as np
import sqlite3
import logging
from pathlib import Path
from datetime import datetime, timedelta
import requests

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(levelname)s %(message)s')
log = logging.getLogger(__name__)

# ── Extract ──────────────────────────────────────────────────
def extract_api_data(endpoint: str, params: dict) -> pd.DataFrame:
    log.info(f'Extracting from {endpoint}')
    r = requests.get(endpoint, params=params, timeout=30)
    r.raise_for_status()
    data = r.json()
    df   = pd.DataFrame(data)
    log.info(f'Extracted {len(df)} rows')
    return df

def extract_csv(filepath: str) -> pd.DataFrame:
    log.info(f'Reading {filepath}')
    df = pd.read_csv(filepath, parse_dates=['date'])
    log.info(f'Read {len(df)} rows, {df.shape[1]} columns')
    return df

# ── Transform ─────────────────────────────────────────────────
def transform_sales(df: pd.DataFrame) -> pd.DataFrame:
    log.info('Transforming sales data')

    # Validate input
    required = ['date', 'product_id', 'quantity', 'unit_price', 'customer_id']
    missing  = [c for c in required if c not in df.columns]
    if missing: raise ValueError(f'Missing columns: {missing}')

    # Clean
    df = df.copy()
    df['date']       = pd.to_datetime(df['date'], errors='coerce')
    df['quantity']   = pd.to_numeric(df['quantity'], errors='coerce').clip(lower=0)
    df['unit_price'] = pd.to_numeric(df['unit_price'], errors='coerce').clip(lower=0)

    # Drop invalid rows
    before = len(df)
    df = df.dropna(subset=['date', 'quantity', 'unit_price'])
    df = df[df['quantity'] > 0]
    log.info(f'Dropped {before - len(df)} invalid rows')

    # Enrich
    df['revenue']      = df['quantity'] * df['unit_price']
    df['year_month']   = df['date'].dt.to_period('M').astype(str)
    df['day_of_week']  = df['date'].dt.day_name()
    df['is_weekend']   = df['date'].dt.dayofweek >= 5
    df['processed_at'] = datetime.utcnow()

    log.info(f'Transform complete: {len(df)} rows, revenue sum = {df["revenue"].sum():,.0f}')
    return df

# ── Load ──────────────────────────────────────────────────────
def load_to_sqlite(df: pd.DataFrame, db_path: str, table: str,
                   if_exists='append'):
    log.info(f'Loading {len(df)} rows to {table}')
    with sqlite3.connect(db_path) as conn:
        df.to_sql(table, conn, if_exists=if_exists, index=False)
    log.info('Load complete')

# ── Orchestrate ───────────────────────────────────────────────
def run_pipeline(date: str):
    log.info(f'=== Starting ETL pipeline for {date} ===')

    # In practice, these come from APIs or file systems
    raw   = extract_csv(f'data/sales_{date}.csv')
    clean = transform_sales(raw)
    load_to_sqlite(clean, 'warehouse.db', 'sales_daily')

    log.info(f'=== Pipeline complete: {len(clean)} rows processed ===')
    return clean

if __name__ == '__main__':
    run_pipeline(datetime.today().strftime('%Y-%m-%d'))

Apache Airflow for Orchestration

pip install apache-airflow

# dags/sales_etl_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash  import BashOperator
from airflow.sensors.filesystem import FileSensor

default_args = {
    'owner':            'dataexpertise',
    'depends_on_past':  False,
    'start_date':       datetime(2026, 9, 1),
    'email_on_failure': True,
    'email':            ['kekare.durgesh9@gmail.com'],
    'retries':          2,
    'retry_delay':      timedelta(minutes=5),
}

with DAG(
    dag_id='sales_etl_daily',
    default_args=default_args,
    description='Daily sales ETL pipeline',
    schedule='0 6 * * *',          # every day at 6am
    catchup=False,
    tags=['etl', 'sales'],
) as dag:

    # Wait for source file to arrive
    wait_for_file = FileSensor(
        task_id='wait_for_sales_file',
        filepath='/data/incoming/sales_{{ ds }}.csv',
        poke_interval=300,   # check every 5 minutes
        timeout=7200,        # fail after 2 hours
        mode='poke'
    )

    # Extract and validate
    def extract(**context):
        from etl.extract import extract_csv
        from etl.validate import validate_schema
        date = context['ds']
        df   = extract_csv(f'/data/incoming/sales_{date}.csv')
        validate_schema(df)
        df.to_parquet(f'/tmp/raw_sales_{date}.parquet', index=False)
        return len(df)

    extract_task = PythonOperator(
        task_id='extract',
        python_callable=extract,
    )

    # Transform
    def transform(**context):
        from etl.transform import transform_sales
        import pandas as pd
        date = context['ds']
        df   = pd.read_parquet(f'/tmp/raw_sales_{date}.parquet')
        clean = transform_sales(df)
        clean.to_parquet(f'/tmp/clean_sales_{date}.parquet', index=False)
        # Push metrics to XCom for downstream tasks
        context['ti'].xcom_push(key='row_count',  value=len(clean))
        context['ti'].xcom_push(key='total_rev',  value=float(clean['revenue'].sum()))

    transform_task = PythonOperator(
        task_id='transform',
        python_callable=transform,
    )

    # Load
    def load(**context):
        from etl.load import load_to_postgres
        import pandas as pd
        date  = context['ds']
        df    = pd.read_parquet(f'/tmp/clean_sales_{date}.parquet')
        load_to_postgres(df, 'sales_daily', date)

    load_task = PythonOperator(
        task_id='load',
        python_callable=load,
    )

    # Run dbt transformations after load
    dbt_run = BashOperator(
        task_id='dbt_run',
        bash_command='cd /dbt && dbt run --select marts.sales_summary --vars '{"date": "{{ ds }}"}''
    )

    dbt_test = BashOperator(
        task_id='dbt_test',
        bash_command='cd /dbt && dbt test --select marts.sales_summary'
    )

    # Task dependencies — define the pipeline graph
    wait_for_file >> extract_task >> transform_task >> load_task >> dbt_run >> dbt_test

Data Quality with Great Expectations

pip install great_expectations

import great_expectations as gx

context = gx.get_context()

# Define expectations for your data
validator = context.sources.pandas_default.read_csv('sales.csv')

validator.expect_column_to_exist('revenue')
validator.expect_column_values_to_not_be_null('date')
validator.expect_column_values_to_not_be_null('customer_id')
validator.expect_column_values_to_be_between('revenue', min_value=0, max_value=1_000_000)
validator.expect_column_values_to_match_regex('date', r'^\d{4}-\d{2}-\d{2}$')
validator.expect_column_values_to_be_unique('transaction_id')
validator.expect_column_median_to_be_between('revenue', min_value=100, max_value=10_000)
validator.expect_table_row_count_to_be_between(min_value=100, max_value=1_000_000)

results = validator.validate()
if not results['success']:
    failed = [r for r in results['results'] if not r['success']]
    print(f'❌ {len(failed)} expectations failed:')
    for f in failed:
        print(f'  - {f["expectation_config"]["expectation_type"]}: '
              f'{f["result"]}')
    raise ValueError('Data quality check failed')
else:
    print('✅ All data quality checks passed')

Conclusion

A robust data pipeline is built in layers: extraction (reliable, idempotent, logged), transformation (tested, reproducible, documented), loading (transactional, schema-validated), and quality (automated checks that fail loudly). Airflow handles orchestration and dependency management. dbt handles SQL transformations with built-in testing. Great Expectations catches data quality issues before they corrupt downstream models. Together they give you the infrastructure that makes data science teams 10x more productive — because reliable data means reliable models.