You pushed the pipeline to production on Friday afternoon. By Monday morning, your Slack is full of angry messages: "The sales dashboard is showing zeros." "Customer records are duplicated." "The revenue numbers look completely wrong."
Sound familiar? ETL pipelines break in predictable ways — but most teams only discover those ways after something blows up in production. This guide shows you the five most common ETL pipeline failure modes and, more importantly, how to catch each one before it reaches your users, using Python and pytest.
The 5 ETL Pipeline Failure Modes
After running ETL pipelines across dozens of projects, the same failure patterns appear again and again. Here they are, ranked by how often they cause production incidents:
| # | Failure Mode | Root Cause | Impact |
|---|---|---|---|
| 1 | Schema drift | Source system changed a column name or type | Pipeline crashes or silently drops data |
| 2 | Null explosions | Required field suddenly contains NULLs | Downstream aggregations return wrong totals |
| 3 | Type mismatches | String "12,500" treated as a number | Calculations fail or silently produce garbage |
| 4 | Duplicate records | Incremental load re-processes already-loaded rows | Revenue totals are inflated; customer counts are wrong |
| 5 | Referential integrity | Foreign key in fact table points to missing dimension row | Joins produce NULL values in reports |
Each of these failures has a corresponding Python test that catches it automatically. Let's build them one by one.
Setting Up Your Test Environment
Install the required libraries:
pip install pytest pandas sqlalchemy pytest-covYour project structure should look like this:
etl_project/
├── pipeline/
│ ├── extract.py
│ ├── transform.py
│ └── load.py
├── tests/
│ ├── conftest.py
│ ├── test_extract.py
│ ├── test_transform.py
│ └── test_load.py
└── requirements.txtCreate your shared test fixtures in conftest.py:
import pytest
import pandas as pd
@pytest.fixture
def sample_source_data():
"""Simulates raw data extracted from source system."""
return pd.DataFrame({
'order_id': [1001, 1002, 1003, 1004],
'customer_id': [101, 102, 103, None],
'amount': ['1,200.50', '850.00', '2,100.75', '300.00'],
'order_date': ['2024-01-15', '2024-01-15', '2024-01-16', '2024-01-16'],
'status': ['completed', 'pending', 'completed', 'cancelled']
})
@pytest.fixture
def expected_schema():
"""Defines the expected column names for the source data."""
return ['order_id', 'customer_id', 'amount', 'order_date', 'status']Failure #1: Schema Drift
The most common silent killer. Your source team renames customer_id to cust_id in an API update. Your pipeline doesn't crash — it just starts loading NULL into the customer column. Nobody notices for two weeks.
import pytest
import pandas as pd
def test_source_schema_has_required_columns(sample_source_data, expected_schema):
"""Fail fast if source columns change."""
actual_columns = list(sample_source_data.columns)
for col in expected_schema:
assert col in actual_columns, (
f"Required column '{col}' is missing from source data. "
f"Got: {actual_columns}"
)
def test_no_unexpected_columns(sample_source_data, expected_schema):
"""Alert when new columns appear — they may need to be mapped."""
actual_columns = set(sample_source_data.columns)
expected_columns = set(expected_schema)
unexpected = actual_columns - expected_columns
assert not unexpected, (
f"Unexpected columns found in source: {unexpected}. "
f"Update your pipeline to handle them."
)
def test_row_count_is_not_zero(sample_source_data):
"""Catch empty extracts — often caused by API failures or wrong date filters."""
assert len(sample_source_data) > 0, "Extract returned 0 rows — check source connection and date filters."Failure #2: Null Explosions
Notice that customer_id for order 1004 is NULL in our sample data. If customer_id is a required field for your fact table, loading this row will either fail on a NOT NULL constraint or silently produce broken joins.
def test_required_fields_have_no_nulls(sample_source_data):
"""Check that fields that must be populated actually are."""
required_non_null = ['order_id', 'amount', 'order_date']
for col in required_non_null:
null_count = sample_source_data[col].isna().sum()
assert null_count == 0, (
f"Column '{col}' has {null_count} NULL value(s). "
f"This field is required to be non-null."
)
def test_nullable_fields_within_acceptable_threshold(sample_source_data):
"""Allow some nulls in optional fields, but alert if too many."""
nullable_col = 'customer_id'
max_null_pct = 0.05 # Allow up to 5% nulls
null_pct = sample_source_data[nullable_col].isna().mean()
assert null_pct <= max_null_pct, (
f"Column '{nullable_col}' has {null_pct:.1%} nulls — "
f"exceeds the {max_null_pct:.0%} threshold. Investigate the source."
)Failure #3: Type Mismatches
Our source data has amount as a string with commas ("1,200.50"). Your transform layer needs to clean this before any arithmetic, or you'll get errors — or worse, silent string concatenation instead of addition.
import pandas as pd
def clean_amount(df: pd.DataFrame) -> pd.DataFrame:
"""Convert amount strings to float, removing commas and currency symbols."""
df = df.copy()
df['amount'] = (
df['amount']
.astype(str)
.str.replace(',', '', regex=False)
.str.replace('$', '', regex=False)
.str.strip()
)
df['amount'] = pd.to_numeric(df['amount'], errors='coerce')
return df
def parse_dates(df: pd.DataFrame) -> pd.DataFrame:
"""Parse order_date to datetime."""
df = df.copy()
df['order_date'] = pd.to_datetime(df['order_date'], format='%Y-%m-%d', errors='coerce')
return dfimport pytest
import pandas as pd
from pipeline.transform import clean_amount, parse_dates
def test_amount_is_numeric_after_transform(sample_source_data):
"""Verify amount column is float after cleaning."""
cleaned = clean_amount(sample_source_data)
assert pd.api.types.is_float_dtype(cleaned['amount']), (
f"Expected float dtype for 'amount', got {cleaned['amount'].dtype}"
)
def test_amount_values_are_positive(sample_source_data):
"""Business rule: order amounts should never be negative."""
cleaned = clean_amount(sample_source_data)
assert (cleaned['amount'] >= 0).all(), "Negative amount values detected after transformation."
def test_date_parsing(sample_source_data):
"""Verify order_date is correctly parsed to datetime."""
parsed = parse_dates(sample_source_data)
assert pd.api.types.is_datetime64_any_dtype(parsed['order_date']), (
"order_date should be datetime64 after transformation."
)
def test_no_nulls_introduced_by_transform(sample_source_data):
"""Ensure transformation doesn't accidentally introduce NULLs into non-null columns."""
original_null_count = sample_source_data['amount'].isna().sum()
cleaned = clean_amount(sample_source_data)
new_null_count = cleaned['amount'].isna().sum()
assert new_null_count == original_null_count, (
f"Transform introduced {new_null_count - original_null_count} new NULL(s) in 'amount'. "
"Check for unparseable values."
)Failure #4: Duplicate Records
Incremental loads are the most common source of duplicates. If your pipeline uses a WHERE updated_at > last_run_time filter, and the previous run failed partway through, you'll re-process rows that were already loaded.
import pytest
import pandas as pd
def test_no_duplicate_order_ids(sample_source_data):
"""Primary key should be unique before loading to warehouse."""
duplicate_ids = sample_source_data[
sample_source_data.duplicated(subset=['order_id'], keep=False)
]['order_id'].tolist()
assert not duplicate_ids, (
f"Duplicate order_id values found: {duplicate_ids}. "
"Use UPSERT logic or dedup before loading."
)
def test_composite_key_uniqueness(sample_source_data):
"""Some tables use composite keys — check those too."""
key_columns = ['order_id', 'order_date']
duplicates = sample_source_data.duplicated(subset=key_columns).sum()
assert duplicates == 0, (
f"Found {duplicates} duplicate row(s) on composite key {key_columns}."
)Failure #5: Referential Integrity
Your orders fact table has a customer_id foreign key that must exist in the customers dimension table. If a customer was deleted from the source but their orders weren't, you'll have orphaned fact rows.
def test_all_customer_ids_exist_in_dimension(sample_source_data):
"""Simulate checking FK relationships before loading."""
# Simulate dimension table
customers_dim = pd.DataFrame({
'customer_id': [101, 102, 103, 104, 105]
})
# Get non-null customer IDs from source
source_customer_ids = set(
sample_source_data['customer_id'].dropna().astype(int).tolist()
)
dim_customer_ids = set(customers_dim['customer_id'].tolist())
orphaned = source_customer_ids - dim_customer_ids
assert not orphaned, (
f"Orphaned customer_id values found: {orphaned}. "
"These orders reference customers that don't exist in the dimension table."
)Running Tests in CI/CD
Tests only protect you if they run automatically on every code change. Here's a minimal GitHub Actions workflow:
name: ETL Pipeline Tests
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: pip install -r requirements.txt
- name: Run ETL tests
run: pytest tests/ -v --cov=pipeline --cov-report=term-missing
- name: Fail if coverage below threshold
run: pytest tests/ --cov=pipeline --cov-fail-under=80With this workflow, every pull request must pass all ETL tests before it can be merged. No more "works on my machine" deployments.
Putting It All Together
Here's your complete test run output with all checks passing:
$ pytest tests/ -v
tests/test_extract.py::test_source_schema_has_required_columns PASSED
tests/test_extract.py::test_no_unexpected_columns PASSED
tests/test_extract.py::test_row_count_is_not_zero PASSED
tests/test_extract.py::test_required_fields_have_no_nulls PASSED
tests/test_extract.py::test_nullable_fields_within_acceptable_threshold PASSED
tests/test_transform.py::test_amount_is_numeric_after_transform PASSED
tests/test_transform.py::test_amount_values_are_positive PASSED
tests/test_transform.py::test_date_parsing PASSED
tests/test_transform.py::test_no_nulls_introduced_by_transform PASSED
tests/test_load.py::test_no_duplicate_order_ids PASSED
tests/test_load.py::test_composite_key_uniqueness PASSED
tests/test_load.py::test_all_customer_ids_exist_in_dimension PASSED
========== 12 passed in 0.42s ==========Frequently Asked Questions
Schema drift occurs when the structure of source data changes unexpectedly — new columns appear, columns are renamed, or data types change — without the ETL pipeline being updated to handle it. This causes downstream failures and corrupted data. The fix is to add automated schema validation tests that run before every pipeline execution.
Use pytest with pandas to write unit tests for each pipeline stage. Test your extract layer for schema consistency and row counts, your transform layer for business logic correctness and type conversions, and your load layer for duplicates and referential integrity. Run these tests in CI/CD so they execute automatically on every code change.
The top 5 ETL pipeline failure causes are: (1) schema drift from upstream sources, (2) null value explosions in required fields, (3) data type mismatches during transformation, (4) duplicate records from incremental loads, and (5) referential integrity violations when loading to the data warehouse. Each has a corresponding Python test that catches it before production.
Absolutely. Integrating pytest-based ETL tests into GitHub Actions (or any CI/CD pipeline) ensures that every code change is validated automatically before deployment. It also protects you from accidental regressions when team members make changes to shared pipeline code.
Asim Noaman Lodhi
Data Engineer & ETL Testing Specialist | Google Partner | 913+ students trained
Author of the Complete ETL Testing Course on Udemy — 79 lectures, 4.5 ★ rating. Covering SQL validation, Python testing, data warehouse QA, and real-world ETL projects.