Building a Reusable ETL Pipeline in pandas: Extract, Transform, and Load Data from Multiple Sources into a Clean, Analysis-Ready Output | Wicked Smart Data
Building a Reusable ETL Pipeline in pandas: Extract, Transform, and Load Data from Multiple Sources into a Clean, Analysis-Ready Output
Learn to build a production-grade ETL pipeline in pandas that extracts data from CSV files, Excel workbooks, and SQL databases; applies a layered transformation strategy; and loads results to multiple output formats. This lesson covers architecture, error handling, validation, and performance — the full picture for data professionals who need pipelines that actually hold up.
🔥 Expert28 min readSep 22, 2026Updated Sep 22, 2026
Building a Reusable ETL Pipeline in pandas: Extract, Transform, and Load Data from Multiple Sources into a Clean, Analysis-Ready Output
Most data work isn't glamorous. Before you write a single aggregation or produce a single chart, you're hunting down files, coaxing them into a consistent shape, fixing someone's inconsistent date formatting, and praying that two datasets actually join on the same key. You do this every month. Sometimes every week. And every time, you start from scratch.
That's the problem an ETL pipeline solves. ETL — Extract, Transform, Load — is a structured approach to moving data from raw sources to an analysis-ready state. The word "pipeline" is key: it implies that each stage feeds the next in a predictable, repeatable way. In production data engineering, ETL pipelines run on orchestration frameworks like Airflow or Dagster. But for the kind of work most data professionals actually do — monthly reports, recurring analysis, consolidating data across departments — a well-structured pandas pipeline is not only sufficient, it's the right tool. You can write it, understand it, and maintain it yourself.
By the end of this lesson, you'll have built a complete, reusable ETL pipeline that extracts data from CSV files, Excel workbooks, and a SQL database; applies a layered transformation strategy that cleans, enriches, and validates the data; and loads the results into both a clean CSV and a formatted Excel workbook. More importantly, you'll understand why the pipeline is structured the way it is, so you can adapt it to your own data problems.
What you'll learn:
How to design a pipeline architecture that separates extraction, transformation, and loading into independent, testable stages
How to write extractor functions that handle real-world messiness at the source level
How to build a layered transformation chain that cleans, standardizes, joins, aggregates, and validates data
How to create a loader that writes to multiple output formats without duplicating logic
How to make the entire pipeline configurable, so changing a file path or output format doesn't require hunting through code
You should have Python 3.9+ installed with pandas, openpyxl, and SQLAlchemy available. If your environment isn't set up yet, see Setting Up Python for Data Analysis.
The Scenario: Monthly Sales Consolidation
Here's the concrete problem we'll solve. Your company tracks sales data across three sources:
A folder of monthly CSV files — each regional sales team exports transactions from their local CRM. Files have slightly different column names and inconsistent date formats.
An Excel workbook — the product catalog, maintained by the merchandising team, with product metadata like category, cost, and target margin.
A SQL database — a customer dimension table containing customer tier, region, and acquisition channel.
Every month, you need to:
Combine all the regional CSV files into one dataset
Join it against the product catalog and customer table
Calculate gross margin and flag orders that fall below the target margin
Validate that no records are missing after the joins
Output a clean CSV for the BI tool and a formatted Excel summary for the finance team
This is a genuine, recurring data problem. Let's build a pipeline that solves it once and runs reliably every time.
Architecture First: Why Structure Matters
Before writing a single line of code, it's worth thinking about structure. The most common mistake in data work is writing a single notebook or script that does everything in sequence — read the files, fix the dates, merge them, calculate something, write the output — all tangled together. This works once. It doesn't scale, doesn't tolerate changes, and is almost impossible to debug when something breaks.
A proper ETL architecture separates three concerns:
┌─────────────────────────────────────────────────────────────────┐
│ EXTRACT │
│ - Read raw data from each source │
│ - Handle source-specific quirks here, not downstream │
│ - Return clean DataFrames with consistent types │
└────────────────────────┬────────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────────┐
│ TRANSFORM │
│ - Standardize column names and values │
│ - Join datasets together │
│ - Derive new columns (margins, flags, categories) │
│ - Validate completeness and business rules │
└────────────────────────┬────────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────────┐
│ LOAD │
│ - Write to output targets (CSV, Excel, database) │
│ - Format and label outputs for the end consumer │
└─────────────────────────────────────────────────────────────────┘
Each stage gets its own functions. Those functions take inputs and return outputs — they don't mutate global state. This means you can test each stage independently, replace any piece without touching the others, and read the pipeline's logic from top to bottom.
Key insight
The single most important architectural decision in an ETL pipeline is making each function pure in spirit — given the same inputs, it produces the same outputs. This makes debugging deterministic. When the pipeline fails, you know exactly which stage to look at.
The project structure we'll build looks like this:
sales_pipeline/
├── config.py # All file paths, column mappings, and parameters
├── extract.py # Extractor functions — one per source type
├── transform.py # Transformation chain functions
├── load.py # Output writers
├── pipeline.py # Orchestrator — calls E, T, and L in order
├── data/
│ ├── raw/ # Source files (CSVs and Excel)
│ └── output/ # Generated outputs
└── run.py # Entry point
The first file we write is config.py. Every file path, column name mapping, database connection string, and output setting lives here. This matters more than it sounds: hardcoded paths inside functions create invisible coupling between your code and your file system. When the path changes — and it will — you have to grep through multiple files to find it.
Using pathlib.Path instead of raw strings means path construction is OS-agnostic and you can use / to join segments cleanly. The SALES_COLUMN_MAP dictionary encodes institutional knowledge about how different teams name the same thing — that's exactly the kind of information that belongs in config, not scattered through transform functions.
The Extract Layer: Source-Specific Readers
The extract layer's job is to read raw data and return DataFrames with consistent types and column names. Anything that's specific to one source — file encoding quirks, sheet names, SQL query logic — lives here.
# extract.py
import pandas as pd
from pathlib import Path
import logging
from config import SALES_CSV_DIR, SALES_COLUMN_MAP, REQUIRED_SALES_COLUMNS
logger = logging.getLogger(__name__)
def _normalize_columns(df: pd.DataFrame, column_map: dict) -> pd.DataFrame:
"""Rename columns using a mapping dict. Only renames columns that exist."""
rename_map = {k: v for k, v in column_map.items() if k in df.columns}
return df.rename(columns=rename_map)
def _validate_required_columns(df: pd.DataFrame, required: list, source: str) -> None:
"""Raise an informative error if any required columns are missing after normalization."""
missing = set(required) - set(df.columns)
if missing:
raise ValueError(
f"Source '{source}' is missing required columns after normalization: {missing}\n"
f"Available columns: {list(df.columns)}"
)
def extract_sales_csvs(csv_dir: Path = SALES_CSV_DIR) -> pd.DataFrame:
"""
Read all CSV files from csv_dir, normalize column names,
and concatenate into a single DataFrame.
Adds a 'source_file' column so you can trace any row back
to its origin — invaluable for debugging join problems.
"""
csv_files = sorted(csv_dir.glob("*.csv"))
if not csv_files:
raise FileNotFoundError(f"No CSV files found in {csv_dir}")
frames = []
for path in csv_files:
logger.info(f"Reading {path.name}")
try:
df = pd.read_csv(
path,
dtype=str, # read everything as string — we'll cast later
encoding="utf-8-sig" # handles Excel-exported CSVs with BOM
)
df = _normalize_columns(df, SALES_COLUMN_MAP)
_validate_required_columns(df, REQUIRED_SALES_COLUMNS, path.name)
df["source_file"] = path.name
frames.append(df)
except Exception as e:
logger.error(f"Failed to read {path.name}: {e}")
raise
combined = pd.concat(frames, ignore_index=True)
logger.info(f"Extracted {len(combined):,} rows from {len(csv_files)} CSV files")
return combined
Two design decisions deserve explanation here.
First, dtype=str on pd.read_csv. Reading everything as a string at extraction time is a deliberate choice. pandas' type inference is convenient but unreliable with real-world data: it will read "00123" as the integer 123, silently drop leading zeros from order IDs, or misparse dates in ambiguous formats like 01/02/23. By deferring all type casting to the transform layer, you maintain full control and avoid silent data corruption.
Second, the source_file column. This is a debugging lifeline. When you discover 47 orders with impossible negative quantities, you can immediately filter to df[df['source_file'] == 'south_region_q3.csv'] and know exactly where to investigate.
Warning
Never concatenate DataFrames inside a loop using df = pd.concat([df, new_df]). This creates a new DataFrame on every iteration and runs in O(n²) time. Always collect DataFrames in a list, then call pd.concat(frames) once at the end.
Extracting from Excel
The product catalog lives in an Excel file with two sheets: Products (the main data) and Discontinued (products no longer active). We want both, but marked differently.
def extract_product_catalog(excel_path: Path = PRODUCT_EXCEL_PATH) -> pd.DataFrame:
"""
Read the active and discontinued product sheets from the Excel catalog.
Marks discontinued products with a flag rather than discarding them —
we still need them to process historical orders.
"""
logger.info(f"Reading product catalog from {excel_path.name}")
xl = pd.ExcelFile(excel_path)
active = pd.read_excel(xl, sheet_name="Products", dtype=str)
active["is_discontinued"] = False
if "Discontinued" in xl.sheet_names:
discontinued = pd.read_excel(xl, sheet_name="Discontinued", dtype=str)
discontinued["is_discontinued"] = True
catalog = pd.concat([active, discontinued], ignore_index=True)
else:
catalog = active
# Normalize column names for the product catalog
catalog.columns = catalog.columns.str.strip().str.lower().str.replace(" ", "_")
logger.info(f"Loaded {len(catalog):,} products ({catalog['is_discontinued'].sum()} discontinued)")
return catalog
Notice how we use pd.ExcelFile to open the workbook once and read multiple sheets, rather than calling pd.read_excel twice. This avoids reading the file from disk twice, which matters when the workbook is large.
Extracting from a SQL Database
The customer dimension table lives in a SQL database. We'll use SQLAlchemy to keep database-specific connection logic out of pandas. For a full treatment of this pattern, see Reading from SQL Databases into pandas with SQLAlchemy.
from sqlalchemy import create_engine, text
from config import CUSTOMER_DB_URL
def extract_customers(db_url: str = CUSTOMER_DB_URL) -> pd.DataFrame:
"""
Pull the customer dimension table from the database.
We SELECT only the columns we need rather than SELECT * —
this keeps the DataFrame small and documents exactly what
we depend on from this source.
"""
logger.info("Connecting to customer database")
engine = create_engine(db_url)
query = text("""
SELECT
customer_id,
customer_name,
customer_tier,
region,
acquisition_channel,
acquisition_date
FROM dim_customers
WHERE is_active = 1
""")
with engine.connect() as conn:
customers = pd.read_sql(query, conn, dtype=str)
logger.info(f"Extracted {len(customers):,} customer records")
return customers
Tip
Always SELECT specific columns from your SQL source, never SELECT *. This makes the pipeline's dependencies explicit, prevents unexpected column additions from breaking downstream transforms, and dramatically reduces memory usage when the table has many columns.
The Transform Layer: A Layered Approach
The transform layer is where most of the complexity lives, and it's where most pipelines become unmaintainable. The solution is to break transformation into a series of small, named functions — each responsible for exactly one kind of change — and then compose them into a chain.
Step 1: Type Casting and Date Parsing
# transform.py
import pandas as pd
import numpy as np
import logging
from config import MARGIN_FLAG_THRESHOLD, LOW_MARGIN_LABEL, OK_MARGIN_LABEL
logger = logging.getLogger(__name__)
def cast_sales_types(df: pd.DataFrame) -> pd.DataFrame:
"""
Cast raw string columns to their correct types.
Coercing with errors='coerce' turns unparseable values into NaN
rather than raising an exception — we catch those NaNs in validation.
"""
df = df.copy()
# Numeric columns
for col in ["quantity", "unit_price"]:
df[col] = pd.to_numeric(df[col], errors="coerce")
# Dates — try ISO format first, then a common US format
# pd.to_datetime with infer_datetime_format handles most real-world cases
df["sale_date"] = pd.to_datetime(df["sale_date"], infer_datetime_format=True, errors="coerce")
# String cleanup — strip whitespace and normalize case for join keys
for col in ["order_id", "customer_id", "product_id"]:
df[col] = df[col].str.strip().str.upper()
return df
The .copy() at the start of each transform function is intentional. Pandas' view/copy semantics are notoriously tricky — modifying a DataFrame you received as an argument can silently modify the caller's data. Calling .copy() at the function boundary eliminates that entire class of bug.
Key insight
The errors='coerce' parameter in pd.to_numeric and pd.to_datetime is one of the most useful patterns in production data work. Instead of crashing on bad data, it converts unparseable values to NaN. This lets you continue processing and then audit the failures in a validation step, rather than fixing one bad row at a time until the pipeline runs.
Step 2: Deduplication
Before joining, we deduplicate on the natural key. Duplicates in source data are common — regional CRMs often export the same order if it was edited after the initial export.
def deduplicate_sales(df: pd.DataFrame) -> pd.DataFrame:
"""
Remove duplicate orders. Keep the last occurrence of each order_id,
on the assumption that later exports reflect more recent edits.
Logs how many duplicates were removed — silence here is a red flag.
"""
original_count = len(df)
df = df.copy()
df = df.sort_values("sale_date").drop_duplicates(
subset=["order_id"], keep="last"
)
removed = original_count - len(df)
if removed > 0:
logger.warning(f"Removed {removed:,} duplicate order_id records")
return df
Step 3: Joining the Datasets
This is the heart of the transformation. We join sales to products and then to customers. The join strategy matters a great deal here.
def join_dimensions(
sales: pd.DataFrame,
products: pd.DataFrame,
customers: pd.DataFrame,
) -> pd.DataFrame:
"""
Join the sales fact table against the product and customer dimension tables.
We use LEFT joins so that every sales row is preserved even if the
dimension lookup fails. Failed lookups will appear as NaN — we catch
them in the validation step rather than silently dropping revenue rows.
"""
df = sales.copy()
# Ensure join keys are consistent types in all DataFrames
for frame, col in [(products, "product_id"), (customers, "customer_id")]:
if col in frame.columns:
frame[col] = frame[col].str.strip().str.upper()
# Join products
df = df.merge(
products[["product_id", "category", "unit_cost", "target_margin", "is_discontinued"]],
on="product_id",
how="left",
indicator=True,
)
unmatched_products = (df["_merge"] == "left_only").sum()
if unmatched_products > 0:
logger.warning(f"{unmatched_products:,} orders could not be matched to a product")
df = df.drop(columns=["_merge"])
# Join customers
df = df.merge(
customers[["customer_id", "customer_name", "customer_tier", "region", "acquisition_channel"]],
on="customer_id",
how="left",
indicator=True,
)
unmatched_customers = (df["_merge"] == "left_only").sum()
if unmatched_customers > 0:
logger.warning(f"{unmatched_customers:,} orders could not be matched to a customer")
df = df.drop(columns=["_merge"])
return df
The indicator=True parameter adds a _merge column that tells you whether each row was matched. Using this, we can count unmatched rows and log a warning without dropping any data. This is a crucial pattern: in an ETL pipeline, silent data loss is far more dangerous than a noisy warning.
def calculate_metrics(df: pd.DataFrame) -> pd.DataFrame:
"""
Add derived business metrics to the joined dataset.
All calculations are vectorized — no apply() or loops.
"""
df = df.copy()
# Cast unit_cost and target_margin to numeric (they came from Excel as strings)
df["unit_cost"] = pd.to_numeric(df["unit_cost"], errors="coerce")
df["target_margin"] = pd.to_numeric(df["target_margin"], errors="coerce")
# Revenue and cost
df["revenue"] = df["quantity"] * df["unit_price"]
df["total_cost"] = df["quantity"] * df["unit_cost"]
# Gross margin — protect against division by zero
df["gross_margin"] = np.where(
df["revenue"] > 0,
(df["revenue"] - df["total_cost"]) / df["revenue"],
np.nan,
)
# Margin flag: compare actual margin against the per-product target
# If no product-level target exists, fall back to the global threshold
effective_threshold = df["target_margin"].fillna(MARGIN_FLAG_THRESHOLD)
df["margin_flag"] = np.where(
df["gross_margin"] < effective_threshold,
LOW_MARGIN_LABEL,
OK_MARGIN_LABEL,
)
# Year-month period for reporting aggregations
df["year_month"] = df["sale_date"].dt.to_period("M").astype(str)
return df
Notice the use of np.where for the conditional columns — this runs as a vectorized C operation rather than a Python loop, which makes an enormous difference on large datasets. If you're still using apply() for this kind of row-wise logic, Writing Fast pandas Code: Vectorization Instead of apply and Loops explains exactly why and how to migrate.
Step 5: Validation
Validation is the step most pipelines skip, and it's the step that would have caught most production incidents. After all transformations are complete, we audit the output for business rule violations and data quality issues.
def validate_output(df: pd.DataFrame) -> pd.DataFrame:
"""
Run data quality checks on the fully transformed DataFrame.
Logs issues as warnings rather than raising exceptions, so the
pipeline completes and the analyst can review the full picture.
For stricter pipelines, raise ValueError instead.
"""
logger.info("Running validation checks...")
# Check 1: No negative quantities or prices
neg_qty = (df["quantity"] < 0).sum()
neg_price = (df["unit_price"] < 0).sum()
if neg_qty > 0:
logger.warning(f"VALIDATION: {neg_qty:,} rows have negative quantity")
if neg_price > 0:
logger.warning(f"VALIDATION: {neg_price:,} rows have negative unit_price")
# Check 2: Missing join results
missing_category = df["category"].isna().sum()
missing_region = df["region"].isna().sum()
if missing_category > 0:
logger.warning(f"VALIDATION: {missing_category:,} rows missing product category (join failed)")
if missing_region > 0:
logger.warning(f"VALIDATION: {missing_region:,} rows missing customer region (join failed)")
# Check 3: Date range sanity — reject future-dated orders
future_orders = (df["sale_date"] > pd.Timestamp.now()).sum()
if future_orders > 0:
logger.warning(f"VALIDATION: {future_orders:,} orders have future sale dates")
# Check 4: Margin calculation completeness
null_margins = df["gross_margin"].isna().sum()
if null_margins > 0:
logger.warning(f"VALIDATION: {null_margins:,} rows have null gross_margin (missing cost data?)")
# Summary
total = len(df)
below_target = (df["margin_flag"] == LOW_MARGIN_LABEL).sum()
logger.info(
f"Validation complete. {total:,} total rows. "
f"{below_target:,} ({below_target/total:.1%}) below margin target."
)
return df # return unchanged — validation is read-only
Composing the Transform Chain
Now we compose all the individual transform functions into a single pipeline function. This is where the architecture pays off — the orchestration is readable as plain English.
The .pipe() method is the idiomatic way to chain DataFrame transformations in pandas. It passes the DataFrame as the first argument to each function, producing a clean functional composition without deeply nested calls. The chain reads like a recipe.
The Load Layer: Writing to Multiple Outputs
The load layer writes the final, validated DataFrame to its output destinations. Notice that load functions should be free of business logic — no calculations, no filtering, no decisions. That all happened upstream. The loader's only job is faithful output.
# load.py
import pandas as pd
from pathlib import Path
import logging
from config import OUTPUT_CSV_PATH, OUTPUT_EXCEL_PATH
logger = logging.getLogger(__name__)
def load_csv(df: pd.DataFrame, path: Path = OUTPUT_CSV_PATH) -> None:
"""Write the clean, analysis-ready dataset to CSV."""
path.parent.mkdir(parents=True, exist_ok=True)
df.to_csv(path, index=False, encoding="utf-8-sig")
logger.info(f"CSV written: {path} ({len(df):,} rows)")
def load_excel_summary(df: pd.DataFrame, path: Path = OUTPUT_EXCEL_PATH) -> None:
"""
Write a summary Excel workbook with two sheets:
1. 'Monthly Summary' — aggregated by year_month and category
2. 'Below Target Orders' — detail rows flagged for low margin
Uses openpyxl engine for formatting. For advanced formatting, see the
openpyxl lesson in this learning path.
"""
path.parent.mkdir(parents=True, exist_ok=True)
# --- Sheet 1: Monthly Summary ---
monthly_summary = (
df.groupby(["year_month", "category"], observed=True)
.agg(
orders=("order_id", "nunique"),
revenue=("revenue", "sum"),
total_cost=("total_cost", "sum"),
avg_margin=("gross_margin", "mean"),
)
.reset_index()
)
monthly_summary["avg_margin"] = monthly_summary["avg_margin"].round(4)
# --- Sheet 2: Below-Target Orders ---
below_target = df[df["margin_flag"] == "BELOW_TARGET"].copy()
below_target_cols = [
"order_id", "sale_date", "customer_name", "product_id",
"category", "region", "revenue", "gross_margin", "source_file"
]
below_target = below_target[[c for c in below_target_cols if c in below_target.columns]]
with pd.ExcelWriter(path, engine="openpyxl") as writer:
monthly_summary.to_excel(writer, sheet_name="Monthly Summary", index=False)
below_target.to_excel(writer, sheet_name="Below Target Orders", index=False)
# Basic column width formatting
for sheet_name, data in [("Monthly Summary", monthly_summary), ("Below Target Orders", below_target)]:
ws = writer.sheets[sheet_name]
for col_idx, column in enumerate(data.columns, 1):
max_len = max(data[column].astype(str).map(len).max(), len(column)) + 2
ws.column_dimensions[ws.cell(row=1, column=col_idx).column_letter].width = min(max_len, 40)
logger.info(f"Excel workbook written: {path}")
# run.py
from pipeline import run_pipeline
run_pipeline()
Separating pipeline.py from run.py means you can also import and call run_pipeline() from a scheduler, a test suite, or a notebook without executing it as a script side effect.
Tip
Logging to both stdout and a file simultaneously gives you real-time visibility during development and a persistent audit trail in production. Set mode="a" on the FileHandler so each run appends rather than overwrites — you'll be glad to have historical run logs when something breaks on the third Tuesday of next month.
Making It Robust: Error Handling Patterns
Production pipelines fail. Files arrive late, database connections time out, source teams introduce new column names without telling anyone. Here are three patterns that make failures informative rather than mysterious.
Pattern 1: Source-Level try/except with Context
Wrap each extractor call with enough context to tell you which file failed and why. A bare except Exception that re-raises is often correct — you want the pipeline to stop on extraction failure, but with a clear error message in the log.
def extract_sales_csvs_robust(csv_dir: Path = SALES_CSV_DIR) -> pd.DataFrame:
csv_files = sorted(csv_dir.glob("*.csv"))
if not csv_files:
raise FileNotFoundError(
f"No CSV files in {csv_dir}. "
f"Confirm the directory path and that files have .csv extension."
)
frames = []
failed = []
for path in csv_files:
try:
df = pd.read_csv(path, dtype=str, encoding="utf-8-sig")
df = _normalize_columns(df, SALES_COLUMN_MAP)
_validate_required_columns(df, REQUIRED_SALES_COLUMNS, path.name)
df["source_file"] = path.name
frames.append(df)
except Exception as e:
logger.error(f"Skipping {path.name}: {e}")
failed.append(path.name)
if failed:
logger.warning(f"The following files were skipped due to errors: {failed}")
if not frames:
raise RuntimeError("All source files failed to load. Pipeline cannot continue.")
return pd.concat(frames, ignore_index=True)
Whether you skip bad files or fail the whole pipeline on one bad file is a business decision. For a financial report where completeness matters, failing hard is the right call. For a monitoring dashboard where partial data is better than no data, the skip-and-log approach makes sense.
Pattern 2: Schema Checkpointing
After extraction, log a schema snapshot: column names, dtypes, row count, and null counts for key columns. This costs almost nothing and is invaluable when debugging intermittent failures.
Call this after each major stage: log_schema_checkpoint(sales_raw, "post-extract"), log_schema_checkpoint(df, "post-join"), and so on.
Pattern 3: Idempotent Outputs
An idempotent pipeline produces the same output when run multiple times with the same inputs. For CSV and Excel outputs, this means overwriting the existing file cleanly. For database outputs, it means using TRUNCATE + INSERT or an UPSERT strategy rather than appending, which creates duplicates on reruns.
def load_to_database(df: pd.DataFrame, engine, table_name: str) -> None:
"""
Write to database using replace strategy — truncates and rewrites
the table on every run, ensuring idempotency.
For incremental strategies, use if_exists='append' with dedup logic.
"""
df.to_sql(
table_name,
engine,
if_exists="replace", # drops and recreates — idempotent
index=False,
chunksize=1000, # write in batches to avoid memory spikes
)
logger.info(f"Loaded {len(df):,} rows to {table_name}")
Hands-On Exercise
Apply what you've built to a realistic scenario. Here's the setup and what you need to implement.
Setup: Create a folder called pipeline_exercise/ with:
data/raw/sales_csv/ — create three small CSV files with intentionally inconsistent column names (use at least two name variations from the SALES_COLUMN_MAP) and a few duplicate order IDs across files.
data/raw/product_catalog.xlsx — an Excel file with a Products sheet and a Discontinued sheet with at least 10 products and two columns: unit_cost and target_margin with per-product variation.
A SQLite database data/raw/customers.db with a dim_customers table (you can create this with SQLAlchemy's df.to_sql() from a small DataFrame).
Tasks:
Extend the extractor to handle a CSV file that uses semicolons as delimiters instead of commas. Add a detect_delimiter helper that tries to infer the delimiter from the first 1KB of the file using Python's csv.Sniffer.
Add a transform step that categorizes orders by revenue size: "Small" (< $500), "Medium" ($500–$5,000), and "Large" (> $5,000). Use pd.cut() for this. See Conditional Column Creation in pandas for the pattern.
Add a third output sheet to the Excel workbook: "By Region" — a pivot showing total revenue and average margin by region and year_month. Use Reshaping Data with pivot_table, melt, and stack to produce the pivot.
Validate the join rate: Add a validation check that raises a RuntimeError if more than 5% of sales rows fail to match a product. This is a hard business rule — a 5% product lookup failure rate implies something fundamentally wrong with the data.
Benchmark the pipeline by running it on 50,000 rows and identifying which stage takes the most time. Use time.perf_counter() around each stage. If the transform stage dominates, consider whether any .apply() calls can be replaced with vectorized operations.
Common Mistakes and Troubleshooting
"My join is creating more rows than the left table"
This almost always means your join key has duplicates in the right-hand table. If product_id appears twice in the products table (once in Products, once in Discontinued), a left join on product_id will create two output rows for every matching sales row. Fix: deduplicate the dimension table on the join key before merging, keeping the most relevant record.
infer_datetime_format=True is convenient but can misparse ambiguous formats. The date 01/02/03 could be January 2, 2003 or February 1, 2003 or February 3, 2001. If your dates are ambiguous, specify the format explicitly:
If different files use different formats, try a list of formats in order:
def parse_date_robustly(series: pd.Series) -> pd.Series:
formats = ["%Y-%m-%d", "%m/%d/%Y", "%d-%m-%Y", "%Y%m%d"]
for fmt in formats:
parsed = pd.to_datetime(series, format=fmt, errors="coerce")
if parsed.notna().sum() > parsed.isna().sum():
return parsed
return pd.to_datetime(series, infer_datetime_format=True, errors="coerce")
"The pipeline runs fine locally but fails in production"
Nine times out of ten, this is a path problem. The script runs from a different working directory in production (cron, Task Scheduler, a CI runner), and relative paths break. Using Path(__file__).parent in config.py anchors all paths to the location of the config file, not the current working directory.
"Memory usage grows every run when running on a schedule"
If your pipeline runs repeatedly in the same process (e.g., called from a long-running scheduler), check for DataFrame references that outlive each run. Python's garbage collector handles most cases, but large DataFrames stored in module-level variables won't be collected. Keep all DataFrames as local variables inside run_pipeline() so they go out of scope and are eligible for collection at the end of each run.
"I can't tell which transformation is corrupting my data"
Add log_schema_checkpoint() calls between every transform step. If that's not enough granularity, write intermediate DataFrames to CSV with df.to_csv(f"debug_{step_name}.csv", index=False) behind a debug flag in config. Once you've identified the problem step, you can call that single function in a notebook with a test DataFrame.
Performance Considerations
For datasets up to a few hundred thousand rows, a well-written pandas pipeline is fast enough that performance isn't a primary concern. But there are a few specific choices that compound into significant time savings at scale.
Read with dtype=str only during extraction. Casting to appropriate types in the transform layer (rather than letting pandas infer during read) is slightly slower at extraction time but prevents expensive re-reads when type inference guesses wrong.
Use categorical dtypes for low-cardinality string columns. Columns like category, region, and customer_tier take far less memory and sort/groupby faster as pd.Categorical:
for col in ["category", "region", "customer_tier", "margin_flag"]:
if col in df.columns:
df[col] = df[col].astype("category")
Minimize the size of dimension tables before joining. If products has 50 columns but you only need 5 for the join and downstream calculations, select those 5 columns before calling merge. Joining wide tables creates wide intermediate DataFrames that consume proportionally more memory.
Profile before optimizing. Use %timeit in Jupyter or cProfile to find the actual bottleneck before rewriting anything. In most pipelines, I/O — reading files, writing Excel — takes more time than any pandas transformation.
Summary and Next Steps
You've built a complete, production-grade ETL pipeline in pandas. Let's review what you implemented and why each decision matters:
config.py centralizes every parameter that might change, so the pipeline can be adapted without touching business logic.
The extract layer handles source-specific messiness at the boundary — column name normalization, encoding, sheet selection — so downstream code doesn't need to defend against it.
dtype=str at read time prevents silent data corruption from type inference, and errors='coerce' lets you process imperfect data and audit failures explicitly.
The transform chain separates each concern — casting, deduplication, joining, enrichment, validation — into composable, testable functions wired together with .pipe().
The load layer contains no business logic — it faithfully writes validated data to multiple output formats.
The orchestrator calls E, T, and L in order with logging and error handling that makes failures diagnosable.
This architecture scales in both directions. You can simplify it for a one-source, one-output pipeline by dropping the modules you don't need. You can extend it by adding new extractors, new transform steps, or new loaders without touching what already works.
If your data includes time series — cumulative metrics, rolling averages, month-over-month comparisons — Working with Dates and Time Series in pandas extends what you built in the metrics step.
If the Excel output needs more sophisticated formatting — conditional colors, charts, print areas — Automating Excel Reports with pandas and openpyxl picks up exactly where the load layer left off.
Before running this pipeline on a new data source for the first time, run the validation and profiling patterns in Validating and Profiling a New Dataset with pandas to understand what you're working with.
The pipeline you've built isn't just a solution to one problem — it's a template. The next time a stakeholder hands you three inconsistent spreadsheets and asks for a monthly report, you'll know exactly where to start.
