Python ETL Patterns for Data Engineers: Extract, Transform, Load Pipelines with pandas, Error Handling, Retry Logic, Logging, Incremental Loading, Data Validation, and Production Architecture

ETL vs ELT — When to Use Each
The Three Phases
Pipeline Architecture: Function-Based
Pipeline Architecture: Class-Based
Error Handling and Retry Logic
Structured Logging for Pipelines
Configuration-Driven Pipelines
Incremental Loading with Watermarks
Data Validation
Idempotent Loads
End-to-End Production Pipeline
Common Mistakes
Interview Questions
Wrapping Up

Our Database Connections post covered how to connect Python to databases. Our File Formats post covered reading and writing CSV, JSON, Parquet, and Excel. This post puts it all together into ETL pipelines — the structured workflows that move data from source systems, clean and transform it, and load it into a destination where it becomes useful.

Analogy — A water treatment plant. Raw water (source data) flows in from rivers and reservoirs (databases, APIs, files). The treatment plant has three stages: intake (extract — pull raw water in), purification (transform — filter sediment, kill bacteria, adjust pH), and distribution (load — push clean water to homes and businesses). If intake fails, nothing flows. If purification is skipped, dirty water reaches consumers. If distribution breaks, clean water sits in tanks doing nothing. Every stage must work, every stage must be monitored, and every failure must be handled gracefully. That is exactly what an ETL pipeline does with data.

ETL vs ELT — When to Use Each

The difference is where the transformation happens. In ETL, you transform data in Python (or Spark) before loading it. In ELT, you load raw data first, then transform it inside the destination (usually with SQL in a data warehouse).

Pattern	Transform Happens	Best For	Tools
ETL	In Python/Spark (before loading)	Complex logic, API data, file processing, cross-source joins	pandas, PySpark, Databricks, custom scripts
ELT	In the warehouse (after loading)	SQL-friendly transforms, large datasets already in the warehouse	dbt, Fabric SQL, Synapse, BigQuery

Most real-world pipelines use both. You might ETL an API response (extract JSON, normalize nested fields in Python, load to staging table), then ELT inside the warehouse (join with dimension tables using SQL). This post focuses on the ETL side — building the Python pipeline that handles extraction, transformation, and loading.

The Three Phases

Phase 1: Extract

Extraction is pulling data from its source — a database, an API, a file on disk, an SFTP server, or a cloud storage bucket. The goal is to get the raw data into Python with as little modification as possible. Do not clean or transform during extraction — that is the next phase.

Analogy — The intake pipe at the water plant. The pipe’s job is to pull water in. It does not filter, it does not purify. It just gets the water from the river to the plant. Mixing intake and purification makes both harder to debug.

import pandas as pd
import requests
from sqlalchemy import create_engine, text
from pathlib import Path

# Extract from a CSV file
def extract_from_csv(filepath, **kwargs):
    # Pull raw data from a CSV file.
    df = pd.read_csv(filepath, **kwargs)
    print(f"Extracted {len(df)} rows from {filepath}")
    return df

# Extract from a database
def extract_from_db(engine, query, params=None):
    # Pull raw data from a database query.
    df = pd.read_sql(text(query), engine, params=params)
    print(f"Extracted {len(df)} rows from database")
    return df

# Extract from a REST API (with pagination)
def extract_from_api(base_url, headers=None, max_pages=100):
    # Pull all pages from a paginated API.
    all_records = []
    page = 1
    while page <= max_pages:
        response = requests.get(f"{base_url}?page={page}", headers=headers, timeout=30)
        response.raise_for_status()
        data = response.json()
        records = data.get("results", [])
        if not records:
            break
        all_records.extend(records)
        page += 1
    df = pd.DataFrame(all_records)
    print(f"Extracted {len(df)} records from API ({page - 1} pages)")
    return df

# Extract from multiple CSV files in a directory
def extract_from_directory(directory, pattern="*.csv"):
    # Combine all matching files into one DataFrame.
    files = sorted(Path(directory).glob(pattern))
    if not files:
        raise FileNotFoundError(f"No files matching {pattern} in {directory}")
    dfs = [pd.read_csv(f) for f in files]
    df = pd.concat(dfs, ignore_index=True)
    print(f"Extracted {len(df)} rows from {len(files)} files")
    return df

Phase 2: Transform

Transformation is where you clean, reshape, enrich, and validate the data. This is the most complex phase — and where most bugs live. Every transform should be a pure function: takes a DataFrame in, returns a DataFrame out, and does not modify the input in place.

Analogy — The purification stage at the water plant. Each filter (transform function) handles one specific contaminant. One removes sediment (null handling), another kills bacteria (deduplication), another adjusts pH (type casting). Separating these into individual steps makes each one testable and debuggable.

import pandas as pd
import numpy as np

def clean_column_names(df):
    # Lowercase, replace spaces with underscores, strip special characters.
    df = df.copy()
    df.columns = (
        df.columns.str.strip()
        .str.lower()
        .str.replace(r"[^a-z0-9]+", "_", regex=True)
        .str.strip("_")
    )
    return df

def drop_empty_rows(df, subset=None):
    # Remove rows where all values (or specified columns) are null.
    before = len(df)
    df = df.dropna(how="all", subset=subset)
    print(f"Dropped {before - len(df)} empty rows")
    return df

def deduplicate(df, subset=None, keep="last"):
    # Remove duplicate rows based on specified columns.
    before = len(df)
    df = df.drop_duplicates(subset=subset, keep=keep)
    print(f"Removed {before - len(df)} duplicates")
    return df

def cast_types(df, type_map):
    # Cast columns to specified types.
    # type_map example: {"order_id": int, "amount": float, "date": "datetime64[ns]"}
    for col, dtype in type_map.items():
        if col in df.columns:
            if dtype == "datetime64[ns]":
                df[col] = pd.to_datetime(df[col], errors="coerce")
            else:
                df[col] = df[col].astype(dtype, errors="ignore")
    return df

def fill_defaults(df, defaults):
    # Fill null values with specified defaults.
    # defaults example: {"status": "unknown", "amount": 0.0}
    return df.fillna(defaults)

def filter_valid_rows(df, conditions):
    # Keep only rows meeting all conditions.
    # conditions: dict of {column: lambda}
    mask = pd.Series(True, index=df.index)
    for col, condition in conditions.items():
        mask = mask & df[col].apply(condition)
    valid = df[mask]
    print(f"Filtered: {len(valid)}/{len(df)} rows passed validation")
    return valid

def add_metadata(df, source_name, load_timestamp):
    # Add audit columns for lineage tracking.
    df = df.copy()
    df["_source"] = source_name
    df["_loaded_at"] = load_timestamp
    return df

Phase 3: Load

Loading writes the transformed data to its destination — a database table, a Parquet file in a data lake, or another system. The key principles are: idempotent (running the load twice produces the same result), atomic (all-or-nothing via transactions), and observable (log what was loaded, how many rows, where).

import pandas as pd
from sqlalchemy import create_engine, text

def load_to_database(df, engine, table_name, schema=None, if_exists="append"):
    # Load a DataFrame into a database table.
    row_count = len(df)
    df.to_sql(
        table_name, engine, index=False,
        if_exists=if_exists, schema=schema,
        chunksize=10_000, method="multi"
    )
    print(f"Loaded {row_count} rows into {schema or 'public'}.{table_name}")

def load_to_parquet(df, output_path, partition_cols=None):
    # Write a DataFrame as partitioned Parquet.
    df.to_parquet(
        output_path, index=False,
        partition_cols=partition_cols,
        compression="snappy"
    )
    print(f"Loaded {len(df)} rows to {output_path}")

def load_with_replace(df, engine, staging_table, target_table, key_columns):
    # Upsert pattern: load to staging, then merge into target.
    # Step 1: Load to staging (replace)
    df.to_sql(staging_table, engine, index=False, if_exists="replace")

    # Step 2: Merge staging into target (upsert)
    key_match = " AND ".join([f"t.{k} = s.{k}" for k in key_columns])
    update_cols = [c for c in df.columns if c not in key_columns]
    update_set = ", ".join([f"t.{c} = s.{c}" for c in update_cols])
    insert_cols = ", ".join(df.columns)
    insert_vals = ", ".join([f"s.{c}" for c in df.columns])

    merge_sql = f"""
        MERGE INTO {target_table} AS t
        USING {staging_table} AS s
        ON {key_match}
        WHEN MATCHED THEN UPDATE SET {update_set}
        WHEN NOT MATCHED THEN INSERT ({insert_cols}) VALUES ({insert_vals});
    """
    with engine.begin() as conn:
        conn.execute(text(merge_sql))
    print(f"Merged {len(df)} rows into {target_table}")

Pipeline Architecture: Function-Based

The simplest pipeline architecture chains extract, transform, and load functions together. Each function is independent and testable. The pipeline function orchestrates the sequence.

Analogy — A factory assembly line. Each station (function) does one thing. The conveyor belt (pipeline function) moves the product (DataFrame) from station to station. If a station breaks, you fix that station without redesigning the entire factory.

from datetime import datetime
import logging

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

def run_orders_pipeline(csv_path, db_engine):
    # End-to-end pipeline: CSV orders to database.
    start = datetime.now()
    logger.info(f"Pipeline started: orders from {csv_path}")

    # EXTRACT
    logger.info("Phase 1: Extract")
    raw = extract_from_csv(csv_path, dtype={"zip_code": str})
    logger.info(f"Extracted {len(raw)} rows")

    # TRANSFORM
    logger.info("Phase 2: Transform")
    df = clean_column_names(raw)
    df = drop_empty_rows(df)
    df = deduplicate(df, subset=["order_id"])
    df = cast_types(df, {"amount": float, "order_date": "datetime64[ns]"})
    df = fill_defaults(df, {"status": "unknown", "amount": 0.0})
    df = add_metadata(df, source_name="vendor_csv", load_timestamp=datetime.now())
    logger.info(f"Transformed: {len(df)} rows after cleaning")

    # LOAD
    logger.info("Phase 3: Load")
    load_to_database(df, db_engine, "orders_clean", if_exists="append")

    duration = (datetime.now() - start).total_seconds()
    logger.info(f"Pipeline completed in {duration:.1f}s ({len(df)} rows)")
    return {"rows": len(df), "duration_seconds": duration}

Pipeline Architecture: Class-Based

For complex pipelines with shared configuration, database connections, and multiple data sources, a class-based architecture keeps everything organized.

import pandas as pd
import logging
from datetime import datetime
from sqlalchemy import create_engine

class ETLPipeline:
    # Base class for ETL pipelines with logging, timing, and error handling.

    def __init__(self, name, db_url=None):
        self.name = name
        self.engine = create_engine(db_url) if db_url else None
        self.logger = logging.getLogger(name)
        self.metrics = {"rows_extracted": 0, "rows_transformed": 0, "rows_loaded": 0}

    def extract(self):
        raise NotImplementedError("Subclass must implement extract()")

    def transform(self, df):
        raise NotImplementedError("Subclass must implement transform()")

    def load(self, df):
        raise NotImplementedError("Subclass must implement load()")

    def validate(self, df, stage):
        # Override for custom validation at each stage.
        if df.empty:
            raise ValueError(f"Validation failed at {stage}: DataFrame is empty")
        return True

    def run(self):
        # Execute the full ETL pipeline with timing and error handling.
        start = datetime.now()
        self.logger.info(f"Pipeline [{self.name}] started")

        try:
            # Extract
            raw = self.extract()
            self.metrics["rows_extracted"] = len(raw)
            self.validate(raw, "extract")

            # Transform
            cleaned = self.transform(raw)
            self.metrics["rows_transformed"] = len(cleaned)
            self.validate(cleaned, "transform")

            # Load
            self.load(cleaned)
            self.metrics["rows_loaded"] = len(cleaned)

            duration = (datetime.now() - start).total_seconds()
            self.logger.info(f"Pipeline [{self.name}] completed in {duration:.1f}s")
            self.logger.info(f"Metrics: {self.metrics}")
            return self.metrics

        except Exception as e:
            self.logger.error(f"Pipeline [{self.name}] failed: {e}", exc_info=True)
            raise


class OrdersPipeline(ETLPipeline):
    # Concrete pipeline: vendor CSV orders to database.

    def __init__(self, csv_path, db_url):
        super().__init__("orders_pipeline", db_url)
        self.csv_path = csv_path

    def extract(self):
        return pd.read_csv(self.csv_path, dtype={"zip_code": str})

    def transform(self, df):
        df.columns = df.columns.str.strip().str.lower().str.replace(r"[^a-z0-9]+", "_", regex=True)
        df = df.drop_duplicates(subset=["order_id"])
        df["amount"] = pd.to_numeric(df["amount"], errors="coerce").fillna(0)
        df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
        df["_loaded_at"] = datetime.now()
        return df

    def load(self, df):
        df.to_sql("orders_clean", self.engine, index=False,
                  if_exists="append", chunksize=10_000, method="multi")


# Usage
pipeline = OrdersPipeline("vendor_data/orders.csv", "postgresql://user:pass@localhost/mydb")
metrics = pipeline.run()

Error Handling and Retry Logic

Production pipelines encounter transient failures — network timeouts, database locks, API rate limits. A robust pipeline retries on transient errors and fails cleanly on permanent errors.

Analogy — A delivery driver. If no one answers the door (transient error), the driver tries again in an hour. If the address does not exist (permanent error), the driver returns the package to the depot. A pipeline should behave the same way.

import time
import logging
from functools import wraps

logger = logging.getLogger(__name__)

def retry(max_attempts=3, backoff_seconds=5, exceptions=(Exception,)):
    # Decorator that retries a function on failure with exponential backoff.
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts:
                        logger.error(f"{func.__name__} failed after {max_attempts} attempts: {e}")
                        raise
                    wait = backoff_seconds * (2 ** (attempt - 1))  # Exponential backoff
                    logger.warning(f"{func.__name__} attempt {attempt} failed: {e}. Retrying in {wait}s...")
                    time.sleep(wait)
        return wrapper
    return decorator


# Apply to extraction functions
@retry(max_attempts=3, backoff_seconds=5, exceptions=(requests.RequestException,))
def extract_from_api(url, headers=None):
    response = requests.get(url, headers=headers, timeout=30)
    response.raise_for_status()
    return pd.DataFrame(response.json()["results"])


@retry(max_attempts=3, backoff_seconds=10)
def load_to_database(df, engine, table_name):
    df.to_sql(table_name, engine, index=False, if_exists="append",
              chunksize=10_000, method="multi")

Dead Letter Pattern

When individual records fail validation or transformation, do not let them kill the entire pipeline. Route bad records to a “dead letter” table or file for manual review later.

def transform_with_dead_letter(df, output_dir="dead_letter/"):
    # Transform valid rows, quarantine bad rows to dead letter file.
    from pathlib import Path
    from datetime import datetime

    # Identify invalid rows
    invalid_mask = (
        df["amount"].isna() |
        (df["amount"] <= 0) |
        df["order_id"].isna() |
        df["order_id"].duplicated(keep="first")
    )

    valid = df[~invalid_mask].copy()
    invalid = df[invalid_mask].copy()

    if len(invalid) > 0:
        dead_letter_dir = Path(output_dir)
        dead_letter_dir.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        dead_letter_path = dead_letter_dir / f"rejected_{timestamp}.csv"
        invalid.to_csv(dead_letter_path, index=False)
        print(f"Quarantined {len(invalid)} bad rows to {dead_letter_path}")

    print(f"Valid: {len(valid)} rows, Invalid: {len(invalid)} rows")
    return valid

Structured Logging for Pipelines

Good logging is the difference between debugging a failure in 5 minutes and debugging it in 5 hours. Use structured logging with consistent fields so you can search and filter logs.

import logging
import json
from datetime import datetime

class PipelineLogger:
    # Structured logger for ETL pipelines.

    def __init__(self, pipeline_name):
        self.pipeline_name = pipeline_name
        self.run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.logger = logging.getLogger(pipeline_name)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter("%(message)s"))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log(self, stage, message, **kwargs):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "pipeline": self.pipeline_name,
            "run_id": self.run_id,
            "stage": stage,
            "message": message,
            **kwargs
        }
        self.logger.info(json.dumps(entry))


# Usage
log = PipelineLogger("orders_etl")
log.log("extract", "Started CSV extraction", file="orders.csv")
log.log("extract", "Extraction complete", rows=15000, duration_ms=450)
log.log("transform", "Deduplication", before=15000, after=14200, removed=800)
log.log("load", "Database insert complete", table="orders_clean", rows=14200)

# Output (structured JSON -- easy to search in Kibana/Splunk/CloudWatch):
# {"timestamp": "2026-07-19T14:30:00", "pipeline": "orders_etl",
#  "run_id": "20260719_143000", "stage": "extract",
#  "message": "Extraction complete", "rows": 15000, "duration_ms": 450}

Configuration-Driven Pipelines

Hardcoding file paths, table names, and connection strings inside your pipeline makes it brittle. Externalize configuration into YAML or JSON files so the same pipeline code works across dev, staging, and production.

import yaml
import os

# config.yaml
# pipeline:
#   name: orders_etl
#   source:
#     type: csv
#     path: /data/vendor_drop/orders.csv
#     dtype:
#       zip_code: str
#   destination:
#     type: database
#     table: orders_clean
#     schema: staging
#     if_exists: append
#   transforms:
#     deduplicate_on: [order_id]
#     required_columns: [order_id, amount, order_date]
#     type_map:
#       amount: float
#       order_date: datetime

def load_config(config_path):
    with open(config_path) as f:
        config = yaml.safe_load(f)
    # Override with environment variables if present
    if os.getenv("DB_URL"):
        config["pipeline"]["destination"]["db_url"] = os.getenv("DB_URL")
    return config["pipeline"]


def run_from_config(config_path):
    config = load_config(config_path)

    # Extract based on source type
    source = config["source"]
    if source["type"] == "csv":
        df = pd.read_csv(source["path"], dtype=source.get("dtype", {}))
    elif source["type"] == "database":
        engine = create_engine(source["db_url"])
        df = pd.read_sql(text(source["query"]), engine)

    # Transform based on config
    transforms = config["transforms"]
    df = df.drop_duplicates(subset=transforms.get("deduplicate_on"))
    # ... apply more transforms from config

    # Load based on destination type
    dest = config["destination"]
    if dest["type"] == "database":
        engine = create_engine(dest["db_url"])
        df.to_sql(dest["table"], engine, index=False,
                  if_exists=dest.get("if_exists", "append"),
                  schema=dest.get("schema"))
    elif dest["type"] == "parquet":
        df.to_parquet(dest["path"], index=False)

Incremental Loading with Watermarks

A full load extracts and loads the entire dataset every time. An incremental load extracts only the rows that changed since the last run. Incremental loading is faster, cheaper, and essential for large datasets.

Analogy — Checking your mailbox. A full load is dumping all your mail from the past year onto the table and reading every piece. An incremental load is checking only the mail that arrived since yesterday. The watermark is the date on yesterday’s last letter.

from sqlalchemy import create_engine, text
from datetime import datetime
import pandas as pd

def get_watermark(engine, pipeline_name):
    # Get the last successful load timestamp for this pipeline.
    with engine.connect() as conn:
        result = conn.execute(
            text("SELECT MAX(last_value) FROM watermarks WHERE pipeline = :name"),
            {"name": pipeline_name}
        )
        row = result.fetchone()
        return row[0] if row and row[0] else datetime(2000, 1, 1)


def update_watermark(engine, pipeline_name, new_value):
    # Update the watermark after a successful load.
    with engine.begin() as conn:
        conn.execute(
            text("""
                INSERT INTO watermarks (pipeline, last_value, updated_at)
                VALUES (:name, :val, :now)
                ON CONFLICT (pipeline)
                DO UPDATE SET last_value = :val, updated_at = :now
            """),
            {"name": pipeline_name, "val": new_value, "now": datetime.now()}
        )


def incremental_extract(source_engine, dest_engine, pipeline_name):
    # Extract only rows modified since the last watermark.
    watermark = get_watermark(dest_engine, pipeline_name)
    print(f"Watermark: {watermark}")

    df = pd.read_sql(
        text("SELECT * FROM orders WHERE modified_at > :wm ORDER BY modified_at"),
        source_engine,
        params={"wm": watermark}
    )
    print(f"Extracted {len(df)} new/modified rows since {watermark}")

    if len(df) > 0:
        new_watermark = df["modified_at"].max()
        # ... transform and load ...
        update_watermark(dest_engine, pipeline_name, new_watermark)
        print(f"Watermark updated to {new_watermark}")

    return df

Data Validation

Validation catches data quality issues before they corrupt your destination tables. Validate after extraction (did we get the right schema?) and after transformation (are the values sensible?).

class DataValidator:
    # Validates a DataFrame against a set of rules.

    def __init__(self, df, name="data"):
        self.df = df
        self.name = name
        self.errors = []

    def not_empty(self):
        if self.df.empty:
            self.errors.append(f"{self.name}: DataFrame is empty")
        return self

    def has_columns(self, columns):
        missing = set(columns) - set(self.df.columns)
        if missing:
            self.errors.append(f"{self.name}: missing columns {missing}")
        return self

    def no_nulls(self, columns):
        for col in columns:
            if col in self.df.columns:
                null_count = self.df[col].isna().sum()
                if null_count > 0:
                    self.errors.append(f"{self.name}: {col} has {null_count} nulls")
        return self

    def unique(self, columns):
        dupes = self.df.duplicated(subset=columns).sum()
        if dupes > 0:
            self.errors.append(f"{self.name}: {dupes} duplicate rows on {columns}")
        return self

    def values_in_range(self, column, min_val=None, max_val=None):
        if column in self.df.columns:
            if min_val is not None and (self.df[column] < min_val).any():
                self.errors.append(f"{self.name}: {column} has values below {min_val}")
            if max_val is not None and (self.df[column] > max_val).any():
                self.errors.append(f"{self.name}: {column} has values above {max_val}")
        return self

    def run(self, raise_on_error=True):
        if self.errors:
            for e in self.errors:
                print(f"  FAIL: {e}")
            if raise_on_error:
                raise ValueError(f"{len(self.errors)} validation errors")
            return False
        print(f"  PASS: {self.name} validation passed")
        return True


# Usage in a pipeline
raw = pd.read_csv("orders.csv")

# Validate after extract
DataValidator(raw, "raw_orders") \
    .not_empty() \
    .has_columns(["order_id", "amount", "order_date"]) \
    .no_nulls(["order_id"]) \
    .run()

# ... transform ...

# Validate after transform
DataValidator(cleaned, "cleaned_orders") \
    .not_empty() \
    .unique(["order_id"]) \
    .values_in_range("amount", min_val=0) \
    .no_nulls(["order_id", "amount", "order_date"]) \
    .run()

Idempotent Loads

An idempotent load produces the same result whether you run it once or five times. This is critical for pipelines that might be retried after a partial failure. The three most common idempotent patterns are: delete-then-insert, upsert (MERGE), and partition overwrite.

# Pattern 1: Delete-then-insert (simplest)
def idempotent_load_delete_insert(df, engine, table, partition_key, partition_value):
    with engine.begin() as conn:
        # Delete existing data for this partition
        conn.execute(
            text(f"DELETE FROM {table} WHERE {partition_key} = :val"),
            {"val": partition_value}
        )
        # Insert fresh data
        df.to_sql(table, conn, index=False, if_exists="append")
    print(f"Idempotent load: deleted and reloaded {len(df)} rows for {partition_key}={partition_value}")


# Pattern 2: Partition overwrite for Parquet
def idempotent_load_parquet(df, base_path, partition_col, partition_value):
    import shutil
    from pathlib import Path
    partition_dir = Path(base_path) / f"{partition_col}={partition_value}"
    if partition_dir.exists():
        shutil.rmtree(partition_dir)  # Remove old partition
    df.to_parquet(str(partition_dir / "data.parquet"), index=False)
    print(f"Overwrote partition {partition_dir}")

End-to-End Production Pipeline

Here is a complete, production-ready pipeline that combines everything from this post: configuration, extraction, transformation, validation, idempotent loading, retry, dead letter handling, and structured logging.

import pandas as pd
import logging
import yaml
import time
from datetime import datetime
from pathlib import Path
from sqlalchemy import create_engine, text

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s [%(name)s] %(levelname)s %(message)s")
logger = logging.getLogger("vendor_orders_etl")


def run_vendor_orders_pipeline(config_path):
    # Production pipeline: vendor CSV to database with full observability.
    start = datetime.now()
    logger.info("=" * 50)
    logger.info("PIPELINE START: vendor_orders_etl")

    # Load config
    with open(config_path) as f:
        config = yaml.safe_load(f)

    source_path = config["source"]["path"]
    db_url = config["destination"]["db_url"]
    target_table = config["destination"]["table"]
    engine = create_engine(db_url)

    # EXTRACT
    logger.info(f"EXTRACT: Reading {source_path}")
    raw = pd.read_csv(source_path, dtype=config["source"].get("dtype", {}))
    logger.info(f"EXTRACT: {len(raw)} rows, {len(raw.columns)} columns")

    # VALIDATE (post-extract)
    required = config["validation"]["required_columns"]
    missing = set(required) - set(raw.columns)
    if missing:
        raise ValueError(f"Missing required columns: {missing}")

    # TRANSFORM
    logger.info("TRANSFORM: Cleaning data")
    df = raw.copy()
    df.columns = df.columns.str.strip().str.lower().str.replace(r"[^a-z0-9]+", "_", regex=True)
    df = df.dropna(how="all")
    df = df.drop_duplicates(subset=config["transforms"]["deduplicate_on"])
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
    df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")

    # DEAD LETTER: quarantine bad rows
    bad_mask = df["amount"].isna() | df["order_id"].isna()
    bad_rows = df[bad_mask]
    df = df[~bad_mask]
    if len(bad_rows) > 0:
        dead_path = Path("dead_letter") / f"rejected_{start.strftime('%Y%m%d_%H%M%S')}.csv"
        dead_path.parent.mkdir(exist_ok=True)
        bad_rows.to_csv(dead_path, index=False)
        logger.warning(f"DEAD LETTER: {len(bad_rows)} rows quarantined to {dead_path}")

    # Add audit metadata
    df["_source_file"] = source_path
    df["_loaded_at"] = datetime.now()

    logger.info(f"TRANSFORM: {len(df)} clean rows ready for load")

    # VALIDATE (post-transform)
    assert len(df) > 0, "No valid rows after transformation"
    assert df["order_id"].is_unique, "Duplicate order_ids found"

    # LOAD (idempotent: delete today's partition, then insert)
    today = datetime.now().strftime("%Y-%m-%d")
    logger.info(f"LOAD: Writing {len(df)} rows to {target_table}")
    with engine.begin() as conn:
        conn.execute(
            text(f"DELETE FROM {target_table} WHERE _source_file = :src"),
            {"src": source_path}
        )
        df.to_sql(target_table, conn, index=False, if_exists="append",
                  chunksize=10_000, method="multi")

    duration = (datetime.now() - start).total_seconds()
    logger.info(f"PIPELINE COMPLETE: {len(df)} rows in {duration:.1f}s")
    logger.info("=" * 50)

    return {
        "rows_extracted": len(raw),
        "rows_loaded": len(df),
        "rows_rejected": len(bad_rows),
        "duration_seconds": round(duration, 1)
    }

Common Mistakes

1. Mixing extraction and transformation. Cleaning data during extraction makes debugging harder because you cannot inspect the raw input. Extract first, then transform. Keep them in separate functions.

2. Modifying DataFrames in place. Functions that modify the input DataFrame with inplace=True create hidden side effects. Always return a new DataFrame (df = df.copy()) so the original stays untouched and each step is independently testable.

3. No logging or metrics. When a pipeline fails at 3 AM, the only question that matters is “where did it fail and what data was it processing?” Without structured logging, you are debugging blind. Log row counts, durations, and file names at every phase.

4. Non-idempotent loads. If your pipeline appends rows without deduplication and gets retried after a partial failure, you end up with duplicate data. Use delete-then-insert, MERGE/upsert, or partition overwrite to ensure reruns produce the same result.

5. Hardcoding connection strings and file paths. Changing from dev to production means editing code. Externalize configuration into YAML files and environment variables so the same code works across environments without modification.

6. No data validation. A missing column, unexpected null, or wrong data type can corrupt your entire destination table. Validate schema after extraction and validate data quality after transformation — before loading anything.

7. Silently dropping bad records. If you dropna() or drop_duplicates() without logging how many rows were removed, you lose visibility into data quality issues. Always log before/after counts, and route bad records to a dead letter file for review.

8. Loading all data as a full load every time. Full loads are simple but wasteful for large tables. A table with 100 million rows that changes 1,000 rows per day should use incremental loading with watermarks. Full loads should be reserved for small tables or periodic full refreshes.

Interview Questions

Q: What is the difference between ETL and ELT, and when would you use each? A: In ETL, data is transformed in an intermediate layer (Python, Spark) before loading into the destination. In ELT, raw data is loaded first, then transformed inside the destination (SQL in a data warehouse). Use ETL when transformations require complex logic that SQL cannot express easily (API pagination, cross-source joins, ML feature engineering). Use ELT when transformations are SQL-friendly and the destination warehouse has abundant compute (Snowflake, BigQuery, Fabric). Most real-world pipelines use both — ETL for ingestion and initial cleaning, ELT for warehouse-side joins and aggregations.

Q: How do you make a pipeline idempotent? A: An idempotent pipeline produces the same result whether run once or multiple times. Three common patterns: delete-then-insert (delete existing records for the partition, then insert fresh data, wrapped in a transaction), upsert with MERGE (match on key columns, update existing rows, insert new ones), and partition overwrite (delete and rewrite the entire date or region partition in a data lake). The key is that a retry after partial failure does not create duplicates.

Q: What is a watermark in the context of incremental loading? A: A watermark is the timestamp or ID of the last successfully processed record. On each run, the pipeline queries only records newer than the watermark, processes them, and updates the watermark to the newest processed value. This avoids re-processing the entire dataset. Watermarks are typically stored in a dedicated table with columns for pipeline name, last value, and last update time.

Q: How do you handle data quality issues in an ETL pipeline? A: Validate at two points: after extraction (schema validation — correct columns, expected types) and after transformation (data quality — no nulls in required columns, values in range, no duplicates on key columns). Route bad records to a dead letter table or file instead of dropping them silently. Log the count and reason for every rejected record. Alert on dead letter volume thresholds.

Q: What is the dead letter pattern and why is it important? A: The dead letter pattern routes records that fail validation or transformation to a separate storage location (dead letter table, CSV file, blob container) instead of letting them crash the pipeline or be silently dropped. This is important because it lets the pipeline continue processing good records while preserving bad ones for investigation. Without dead letter handling, one malformed record can halt the entire pipeline, or you lose visibility into data quality issues.

Q: How would you design a pipeline that processes data from three different sources? A: Extract from each source independently into separate DataFrames (parallel extraction if possible). Validate each DataFrame against its expected schema. Transform each source separately (different cleaning rules per source). Join or union the transformed DataFrames into a single dataset. Validate the combined dataset. Load to the destination. The key architectural principle is that each source extraction and transformation is independent — a failure in one source should not prevent the other two from loading.

Q: Why should ETL functions be pure functions that return new DataFrames instead of modifying in place? A: Pure functions (input DataFrame in, new DataFrame out) are independently testable — you can unit test each transform with a small test DataFrame. They are composable — you can chain them in any order or skip a step. They are debuggable — you can inspect the DataFrame before and after each step. In-place modifications create hidden dependencies between steps, making it impossible to test a single transform in isolation or reorder the pipeline without side effects.

Wrapping Up

ETL pipelines are the backbone of data engineering. The patterns in this post — function-based pipelines, retry with exponential backoff, dead letter handling, structured logging, configuration-driven architecture, incremental loading with watermarks, data validation, and idempotent loads — apply whether you are building a simple CSV-to-database script or a multi-source enterprise pipeline. Start simple (function-based), add observability (logging and validation), then add robustness (retry, dead letter, idempotency). Every production pipeline needs all three layers.

← Previous: Database Connections Python (24/31) Next: Testing with pytest →

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.

Python ETL Patterns for Data Engineers: Extract, Transform, Load Pipelines with pandas, Error Handling, Retry Logic, Logging, Incremental Loading, Data Validation, and Production Architecture

Table of Contents

ETL vs ELT — When to Use Each

The Three Phases

Phase 1: Extract

Phase 2: Transform

Phase 3: Load

Pipeline Architecture: Function-Based

Pipeline Architecture: Class-Based

Error Handling and Retry Logic

Dead Letter Pattern

Structured Logging for Pipelines

Configuration-Driven Pipelines

Incremental Loading with Watermarks

Data Validation

Idempotent Loads

End-to-End Production Pipeline

Common Mistakes

Interview Questions

Wrapping Up

Leave a Comment Cancel Reply

Table of Contents

ETL vs ELT — When to Use Each

The Three Phases

Phase 1: Extract

Phase 2: Transform

Phase 3: Load

Pipeline Architecture: Function-Based

Pipeline Architecture: Class-Based

Error Handling and Retry Logic

Dead Letter Pattern

Structured Logging for Pipelines

Configuration-Driven Pipelines

Incremental Loading with Watermarks

Data Validation

Idempotent Loads

End-to-End Production Pipeline

Common Mistakes

Interview Questions

Wrapping Up

Related Posts

Leave a Comment Cancel Reply