Build an ETL Pipeline with pandas and Parquet

Overview

pandas is the standard tool for tabular data processing in Python. Parquet is a columnar storage format that compresses better than CSV and preserves data types (integers, floats, datetimes, categoricals). Combining them in an ETL pipeline gives you type-safe data processing with compact storage. The examples below show how to extract from multiple sources, transform with type coercion and validation, and load to partitioned Parquet files.

When to Use

• Batch data processing jobs that run on a schedule (hourly, daily)
• Transforming CSV/JSON exports into typed Parquet for downstream analytics
• Data pipelines where intermediate files need type preservation
• Building features for ML models from raw data sources
• Any scenario where you need reproducible, auditable data transformations

When NOT to Use

• Streaming/real-time pipelines — use Spark Structured Streaming or Flink
• Datasets larger than memory — use Polars, Dask, or PySpark instead
• Simple one-off transformations — a single pd.read_csv().to_parquet() is enough
• Production data warehouses — use dbt for SQL-based transformations

Solution

The snippets below are copy-paste ready. A single runnable version that generates sample data and runs the whole pipeline end-to-end lives in the companion repository.

Basic ETL pipeline structure

import pandas as pd
from pathlib import Path
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def extract_csv(path: str) -> pd.DataFrame:
    """Extract data from a CSV file."""
    logger.info(f"Extracting from {path}")
    df = pd.read_csv(path)
    logger.info(f"Extracted {len(df)} rows, {len(df.columns)} columns")
    return df


def extract_json(path: str) -> pd.DataFrame:
    """Extract data from a JSON file."""
    logger.info(f"Extracting from {path}")
    df = pd.read_json(path, lines=True)
    logger.info(f"Extracted {len(df)} rows")
    return df


def transform(df: pd.DataFrame) -> pd.DataFrame:
    """Apply transformations: type coercion, cleaning, derived columns."""
    logger.info("Starting transformation")

    # Type coercion
    df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
    df["quantity"] = pd.to_numeric(df["quantity"], errors="coerce").astype("Int64")

    # Drop rows with invalid dates or amounts
    df = df.dropna(subset=["order_date", "amount"])

    # Derive columns
    df["year"] = df["order_date"].dt.year
    df["month"] = df["order_date"].dt.month
    df["revenue"] = df["amount"] * df["quantity"]

    # Normalize text columns
    df["customer_name"] = df["customer_name"].str.strip().str.title()

    # Categorical for low-cardinality columns
    df["status"] = df["status"].astype("category")

    logger.info(f"Transformed to {len(df)} rows")
    return df


def load_parquet(df: pd.DataFrame, path: str, partition_cols: list[str] | None = None) -> None:
    """Load DataFrame to Parquet, optionally partitioned."""
    logger.info(f"Loading to {path}")
    if partition_cols:
        df.to_parquet(path, partition_cols=partition_cols, index=False)
    else:
        df.to_parquet(path, index=False)
    logger.info(f"Loaded {len(df)} rows")


def run_pipeline(source_path: str, destination_path: str) -> None:
    """Run the full ETL pipeline."""
    df = extract_csv(source_path)
    df = transform(df)
    load_parquet(df, destination_path, partition_cols=["year", "month"])


if __name__ == "__main__":
    run_pipeline("data/raw/orders.csv", "data/processed/orders")

Extract from multiple sources and merge

def extract_and_merge(orders_path: str, customers_path: str) -> pd.DataFrame:
    """Extract from multiple sources and merge."""
    orders = pd.read_csv(orders_path)
    customers = pd.read_csv(customers_path)

    # Standardize join keys
    orders["customer_id"] = orders["customer_id"].astype(str).str.strip()
    customers["customer_id"] = customers["customer_id"].astype(str).str.strip()

    merged = orders.merge(customers, on="customer_id", how="left")

    logger.info(f"Merged: {len(orders)} orders + {len(customers)} customers = {len(merged)} rows")
    return merged

Transform with validation

def transform_with_validation(df: pd.DataFrame) -> pd.DataFrame:
    """Transform with data quality checks."""
    # Type coercion
    df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

    # Validation: no negative amounts
    negative_count = (df["amount"] < 0).sum()
    if negative_count > 0:
        logger.warning(f"Found {negative_count} negative amounts, filtering out")
        df = df[df["amount"] >= 0]

    # Validation: no duplicate order IDs
    dup_count = df.duplicated(subset=["order_id"]).sum()
    if dup_count > 0:
        logger.warning(f"Found {dup_count} duplicate order IDs, dropping duplicates")
        df = df.drop_duplicates(subset=["order_id"], keep="last")

    # Validation: required columns present
    required_cols = ["order_id", "customer_id", "order_date", "amount"]
    missing = [c for c in required_cols if c not in df.columns]
    if missing:
        raise ValueError(f"Missing required columns: {missing}")

    # Derived columns
    df["year"] = df["order_date"].dt.year
    df["month"] = df["order_date"].dt.month
    df["quarter"] = df["order_date"].dt.quarter

    return df

Partitioned Parquet output

def load_partitioned(df: pd.DataFrame, base_path: str) -> None:
    """Load to partitioned Parquet by year and month."""
    # Ensure partition columns are strings (Parquet requirement)
    df["year"] = df["year"].astype(str)
    df["month"] = df["month"].astype(str).str.zfill(2)

    df.to_parquet(
        base_path,
        partition_cols=["year", "month"],
        index=False,
        engine="pyarrow",
        compression="snappy",
    )
    logger.info(f"Partitioned output at {base_path}/year=*/month=*")


def read_partitioned(base_path: str, year: str, month: str | None = None) -> pd.DataFrame:
    """Read specific partitions."""
    if month:
        path = f"{base_path}/year={year}/month={month}"
    else:
        path = f"{base_path}/year={year}"
    return pd.read_parquet(path)

Incremental load (append to existing Parquet)

def load_incremental(df: pd.DataFrame, path: str) -> None:
    """Append new data to existing Parquet dataset."""
    from pathlib import Path

    if Path(path).exists():
        existing = pd.read_parquet(path)
        combined = pd.concat([existing, df], ignore_index=True)
        combined = combined.drop_duplicates(subset=["order_id"], keep="last")
    else:
        combined = df

    combined.to_parquet(path, index=False)
    logger.info(f"Incremental load: {len(df)} new rows, {len(combined)} total")

Pipeline with error handling and retries

import time

def extract_with_retry(path: str, retries: int = 3, delay: int = 5) -> pd.DataFrame:
    """Extract with retry logic for network sources."""
    for attempt in range(retries):
        try:
            if path.startswith("http"):
                df = pd.read_csv(path)
            else:
                df = pd.read_csv(path)
            return df
        except Exception as e:
            logger.warning(f"Attempt {attempt + 1}/{retries} failed: {e}")
            if attempt < retries - 1:
                time.sleep(delay * (attempt + 1))
            raise


def run_pipeline_safe(source: str, destination: str) -> bool:
    """Run pipeline with full error handling."""
    try:
        df = extract_with_retry(source)
        df = transform_with_validation(df)
        load_partitioned(df, destination)
        logger.info("Pipeline completed successfully")
        return True
    except Exception as e:
        logger.error(f"Pipeline failed: {e}")
        return False

Schema enforcement

EXPECTED_SCHEMA = {
    "order_id": "int64",
    "customer_id": "object",
    "order_date": "datetime64[ns]",
    "amount": "float64",
    "quantity": "Int64",
    "status": "category",
}

def enforce_schema(df: pd.DataFrame) -> pd.DataFrame:
    """Enforce expected schema on DataFrame."""
    for col, dtype in EXPECTED_SCHEMA.items():
        if col not in df.columns:
            raise ValueError(f"Missing column: {col}")
        if df[col].dtype != dtype:
            logger.info(f"Converting {col} from {df[col].dtype} to {dtype}")
            if dtype == "datetime64[ns]":
                df[col] = pd.to_datetime(df[col], errors="coerce")
            elif dtype == "category":
                df[col] = df[col].astype("category")
            else:
                df[col] = df[col].astype(dtype)
    return df

Explanation

The pipeline is a linear graph with a failure path: each stage consumes the previous stage’s output and logs row counts, so you can see exactly where records were dropped.

Why Int64 instead of int64: errors="coerce" turns unparseable values into NaN, and the plain int64 dtype can’t hold NaN. The cast itself would raise. The nullable Int64 dtype stores pd.NA instead, so quantity survives rows whose raw value wasn’t numeric.

Why partition by year/month: partition_cols writes a directory layout (year=2026/month=10/) that query engines use for partition pruning — a query filtered on one month reads only that folder’s files instead of scanning the whole dataset. Keep granularity at the level your queries actually filter by: daily partitions on small data produce thousands of tiny files, which ends up slower than one medium file.

Which compression codec:

Codec	Speed	Ratio	Pick it when
snappy	Fastest	Lowest	Files are read often — the default for a reason
zstd	Fast	~2x better than snappy	Storage-bound or archival datasets
gzip	Slowest	Good	A downstream tool can’t read snappy/zstd

Retries belong to I/O, not transforms: a transform that fails on bad data fails on every retry, while a flaky filesystem or network source often succeeds on the second attempt. That’s why the retry wrapper wraps extract only: transform errors surface immediately instead of burning through retries.

For declarative schema rules and column-level checks beyond what transform_with_validation does here, see Validate pandas DataFrames with Pandera.

Variants

Using PyArrow directly for large files

import pyarrow.parquet as pq
import pyarrow as pa

def load_with_pyarrow(df: pd.DataFrame, path: str) -> None:
    """Write Parquet using PyArrow for more control."""
    table = pa.Table.from_pandas(df, preserve_index=False)
    pq.write_table(
        table,
        path,
        compression="zstd",
        compression_level=3,
        use_dictionary=True,
        write_statistics=True,
    )

Pipeline with logging to file

import logging.handlers

def setup_logging(log_path: str = "logs/etl.log") -> None:
    """Set up file + console logging."""
    handler = logging.handlers.RotatingFileHandler(
        log_path, maxBytes=10_000_000, backupCount=5
    )
    handler.setFormatter(logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    ))
    logging.basicConfig(
        level=logging.INFO,
        handlers=[handler, logging.StreamHandler()],
    )

Pipeline orchestration with config

import yaml

def load_config(config_path: str) -> dict:
    with open(config_path) as f:
        return yaml.safe_load(f)

def run_pipeline_from_config(config_path: str) -> None:
    config = load_config(config_path)

    df = pd.read_csv(config["source"])
    for transform_config in config.get("transforms", []):
        if transform_config["type"] == "rename":
            df = df.rename(columns=transform_config["mapping"])
        elif transform_config["type"] == "filter":
            df = df.query(transform_config["condition"])
        elif transform_config["type"] == "cast":
            df[transform_config["column"]] = df[transform_config["column"]].astype(
                transform_config["dtype"]
            )

    df.to_parquet(
        config["destination"],
        partition_cols=config.get("partition_cols"),
        index=False,
    )

Best Practices

• For a deeper guide, see Schedule and Monitor DAGs with Apache Airflow.

• Use errors="coerce" in pd.to_numeric and pd.to_datetime: a malformed value becomes NaN you can count and drop, instead of an exception that kills the whole run

• Partition by date columns (year, month) — readers can prune to specific time ranges instead of scanning every file

• Use snappy compression for speed, zstd for better compression ratio — pick per the codec table above

• Log row counts at each stage: the log tells you which stage dropped rows without any extra debugging

• Validate data before writing — a bad row caught in transform is cheap; the same row in Parquet poisons every downstream read

• Use Int64 (nullable integer) instead of int64 when data may have missing values — int64 can’t represent NaN and the cast fails mid-pipeline

• Write statistics in Parquet — min/max metadata lets readers use predicate pushdown and skip row groups that can’t match

Common Mistakes

• Not handling missing values: pd.to_numeric without errors="coerce" raises on the first invalid string it hits — often deep into a large file. Coerce first, then decide whether to drop or fill the resulting NaNs.
• Using CSV as intermediate format: CSV stores everything as text, so your careful dtypes are gone on the next read. Parquet preserves them.
• Not partitioning large datasets: a single 10GB Parquet file forces every query to read everything. Partition by the column your queries filter on most.
• Ignoring dtypes after reading: pd.read_csv infers types and often guesses wrong — IDs become floats, zip codes lose leading zeros. Cast the columns you care about explicitly.
• Not deduplicating on incremental loads: re-running a failed job appends the same batch twice. Deduplicate on a natural key like order_id before writing.

Further Reading

• pandas to_parquet reference — partition_cols, compression, and engine options
• PyArrow parquet module — row groups, statistics, and low-level write control
• Apache Parquet format docs — encodings, compression, and file layout
• pandas IO tools: partitioning datasets — how pandas handles key=value directory layouts