Documented Cleaning Pipelines

From Ad-Hoc Queries to Scripted Pipelines

Ad-hoc cleaning is writing a query, running it, eyeballing the result, and writing another query to fix what the first one missed. It works for a one-time exploration. It fails for any analysis that will be repeated, shared, or audited.

The problem with ad-hoc cleaning is not that it produces wrong results. It produces results that cannot be reproduced. Six months later, when someone asks how a number was calculated, the analyst who wrote the queries has left the company. The queries were never saved. The intermediate results were overwritten. The answer is unknowable.

What a Pipeline Is

A cleaning pipeline is a scripted sequence of transformations that converts raw data into analysis-ready data. Each step is explicit, ordered, and documented. Given the same raw input, the pipeline always produces the same clean output.

raw_data/
  ├── customers_2024_01.csv
  └── orders_2024_01.csv

cleaning_pipeline/
  ├── 01_validate_raw.py
  ├── 02_normalize_strings.py
  ├── 03_parse_dates.py
  ├── 04_deduplicate.py
  └── 05_validate_output.py

data_dictionary.md
cleaning_log.md

Each script performs one category of transformation. The numbering enforces execution order. The pipeline can be run by anyone, at any time, with the same result.

The Three-Layer Architecture

A well-designed pipeline separates data into three layers:

Raw layer: The original data exactly as received. Never modified. This is your ground truth. If a cleaning step destroys information, you can always return to the raw layer and start over.

Staging layer: Data that has been parsed and loaded but not yet cleaned. Types are converted, files are merged, and basic structural issues are resolved. The staging layer is the input to the cleaning steps.

Clean layer: Analysis-ready data. All cleaning steps have been applied. All transformations have been documented. This is the layer that feeds reports, dashboards, and models.

import pandas as pd

raw = pd.read_csv('raw_data/orders_2024_01.csv')

staging = raw.copy()
staging['order_date'] = pd.to_datetime(staging['order_date'], format='mixed', errors='coerce')
staging['revenue'] = pd.to_numeric(staging['revenue'].str.replace('$', '', regex=False).str.replace(',', '', regex=False), errors='coerce')

clean = staging.dropna(subset=['customer_id', 'order_date', 'revenue'])
clean = clean.drop_duplicates(subset=['order_id'], keep='last')
clean.to_parquet('clean/orders_clean.parquet', index=False)

This script reads raw data, converts types in the staging phase, applies cleaning rules, and writes the clean output. The raw file is untouched. The clean file is fully derived from documented steps.

Why Scripting Beats GUI Tools

GUI tools like Excel or Power Query are useful for exploration. They are unacceptable for production pipelines because:

1. Not reproducible: A colleague cannot re-run your Excel steps without the original file and your click sequence.
2. Not versionable: There is no diff between "the pipeline I used last month" and "the pipeline I use now."
3. Not testable: You cannot write automated tests for a series of mouse clicks.
4. Not schedulable: A script can run on a schedule; a GUI workflow requires a human.

A scripted pipeline solves all four problems. It is text, so it can be version-controlled. It is deterministic, so it can be tested. It is executable, so it can be scheduled.

Pipeline Design Principles

A well-designed pipeline follows these principles:

Single responsibility: Each script does one thing. 01_validate_raw.py validates. 02_normalize_strings.py normalizes. 03_parse_dates.py parses dates. This separation makes each step testable and replaceable.

Fail fast: If the raw data fails validation, stop immediately. Do not attempt to clean data that is structurally broken. A pipeline that produces output from invalid input gives false confidence.

Idempotent: Running the pipeline twice on the same input produces the same output. This property is essential for recovery. If a pipeline fails halfway through, you can re-run it from the start without worrying about partial state.

Observable: Every run produces metrics: rows in, rows out, rows quarantined, validation results. These metrics are written to a log and monitored over time. A pipeline you cannot observe is a pipeline you cannot trust.

Configurable: Parameters (date formats, thresholds, vocabularies) live in configuration files, not in code. Changing a parameter should not require editing a script. This separation allows the same pipeline to process different datasets with different characteristics.

Documenting Every Transformation

Documentation is not a deliverable you add after the pipeline works. It is a structural component of the pipeline itself. A pipeline without documentation is a black box: data goes in, results come out, and nobody knows what happened in between.

The Data Dictionary

A data dictionary describes every column in the clean dataset: its name, type, meaning, allowed values, and the transformations applied to produce it.

# Data Dictionary: orders_clean

| Column | Type | Description | Transformations |
|--------|------|-------------|------------------|
| order_id | STRING | Unique order identifier | Trimmed whitespace; validated non-null |
| customer_id | INTEGER | Customer FK; references customers.customer_id | Cast from text; leading zeros preserved in staging |
| order_date | DATE | Date the order was placed | Parsed from mixed formats (ISO, US, European) |
| revenue | DECIMAL(12,2) | Order total in USD | Removed $ and commas; converted to numeric |
| is_outlier | BOOLEAN | True if revenue exceeds IQR upper bound | Computed via 1.5 * IQR rule |
| region | STRING | Customer region | Normalized to controlled vocabulary (NA, EU, APAC, LATAM) |

The dictionary serves two audiences. Analysts use it to understand what each column means before writing queries. Auditors use it to verify that the documented transformations match what the pipeline actually does.

The Cleaning Log

The cleaning log records what the pipeline did to the data. It is structured like a commit message: dated, specific, and focused on what changed and why.

# Cleaning Log: orders pipeline

## 2024-01-15
- Parsed order_date from mixed formats; 342 rows returned NULL and were quarantined
- Removed $ and comma from revenue; cast to DECIMAL(12,2)
- Deduplicated on order_id keeping most recent by updated_at; removed 12 duplicate rows
- Flagged 8 revenue outliers using IQR method (upper bound: $4,250.00)
- Validated: 0 NULL customer_id, 0 NULL order_date, 0 negative revenue

## 2024-01-08
- Initial pipeline version
- Loaded 12,450 raw orders from CSV
- Applied string normalization to region column

Each entry answers: What was done? How many rows were affected? What was the result of the validation checks? This log is the audit trail that makes the pipeline trustworthy.

Quarantine, Don't Destroy

When a row fails validation, do not silently delete it. Move it to a quarantine table with the reason for rejection. This preserves the data for investigation and prevents irreversible loss.

CREATE TABLE quarantine_orders AS
SELECT *, 'null_customer_id' AS rejection_reason
FROM staging_orders
WHERE customer_id IS NULL;

CREATE TABLE quarantine_orders_2 AS
SELECT *, 'future_date' AS rejection_reason
FROM staging_orders
WHERE order_date > CURRENT_DATE;

Quarantined rows can be fixed and reprocessed, or they can reveal systematic problems in the upstream data source. Either way, the information is preserved rather than discarded.

Decision Records

For significant cleaning decisions, write a brief decision record that explains the alternatives considered and why the chosen approach was selected.

## ADR-003: Outlier handling for revenue

**Context:** Revenue column contains values up to $2.3M, which distort mean-based metrics.

**Decision:** Flag outliers using the IQR method; do not remove or cap.

**Alternatives considered:**
- Removal: rejected because high-value orders are legitimate enterprise sales
- Capping (Winsorization): rejected because it would mask true revenue variance
- Z-score: rejected because revenue distribution is heavily right-skewed

**Consequences:** Downstream analyses must filter or stratify by is_outlier flag.

Decision records prevent the "why did we do this?" question from arising months later. They also prevent the same debate from being re-litigated every time a new team member joins.

Version Control for Data Pipelines

Version control is not just for application code. Cleaning logic changes over time, and those changes must be tracked with the same rigor as software changes. A pipeline that cannot be rolled back to a previous version is a pipeline that cannot be trusted.

What to Version

Everything that affects the clean output must be version-controlled:

• Cleaning scripts: The SQL or Python code that performs transformations
• Configuration files: Date format strings, IQR multipliers, controlled vocabularies
• Mapping tables: Category mappings, unit conversion factors
• Data dictionaries: Column definitions and transformation descriptions
• Tests: Validation queries and expected results

What should not be version-controlled: raw data (too large, already preserved by the source system), clean data (derived, reproducible by running the pipeline), and credentials (use environment variables or a secrets manager).

Semantic Versioning for Pipelines

Use semantic versioning to communicate the nature of changes:

• Major (1.0.0 → 2.0.0): Breaking change. Column renamed, removed, or redefined. Existing queries will break.
• Minor (1.0.0 → 1.1.0): New column added, new validation rule, or new quarantine category. Existing queries continue to work.
• Patch (1.0.0 → 1.0.1): Bug fix, typo correction, or performance improvement. No change to output schema.

cleaning_pipeline/
  ├── v1.0.0/
  │   ├── pipeline.py
  │   ├── config.yaml
  │   └── data_dictionary.md
  ├── v1.1.0/
  │   ├── pipeline.py
  │   ├── config.yaml
  │   └── data_dictionary.md
  └── v1.1.1/
      ├── pipeline.py
      ├── config.yaml
      └── data_dictionary.md

Branching Strategy for Data Work

Use branches to develop and test changes without affecting the production pipeline:

1. Main branch: The production pipeline. Always works, always tested.
2. Feature branches: Develop new cleaning rules or fix bugs. Test against a sample of raw data.
3. Pull request: Review the diff, run the full test suite, and merge only when all checks pass.

git checkout -b fix-date-parsing
# Edit pipeline.py to handle new date format
python test_pipeline.py
# All tests pass
git commit -m "Handle DD/MM/YYYY date format in order_date column"
git push origin fix-date-parsing
# Create pull request, review, merge

Reproducibility Through Pinning

A pipeline that produces different results each time it runs is not a pipeline; it is a random number generator. Ensure reproducibility by pinning:

• Library versions: Use a requirements.txt or environment.yml with exact versions
• Data snapshots: Record the hash or timestamp of the raw data used
• Random seeds: Set seeds for any stochastic operations (e.g., train/test splits)
• Configuration: Store all parameters in a versioned config file, not hardcoded in scripts

# config.yaml
pipeline_version: 1.2.0
raw_data_snapshot: orders_2024_01_v3.sha256
outlier_method: iqr
outlier_multiplier: 1.5
date_formats:
  - YYYY-MM-DD
  - MM/DD/YYYY
quarantine_threshold: 0.05

When a stakeholder asks, "Can you reproduce the Q3 report?" the answer should be: "Run the v1.2.0 pipeline against the Q3 raw snapshot." That is a reproducible analysis.

Rollback and Recovery

Version control enables rollback, but rollback is only useful if you can recover the state that produced a given output. Store the following for every pipeline run:

• Input hash: A checksum of the raw data file. This verifies that the input has not changed.
• Pipeline version: The version of the cleaning code that was executed.
• Configuration snapshot: The exact configuration used for the run.
• Output hash: A checksum of the clean output. This verifies that the output has not been modified since creation.

import hashlib

def file_hash(path):
    h = hashlib.sha256()
    with open(path, 'rb') as f:
        for chunk in iter(lambda: f.read(8192), b''):
            h.update(chunk)
    return h.hexdigest()

run_record = {
    'run_date': '2024-01-15T09:30:00Z',
    'pipeline_version': '1.2.0',
    'input_hash': file_hash('raw/orders_2024_01.csv'),
    'output_hash': file_hash('clean/orders_clean.parquet'),
    'config': 'config.yaml'
}

With these records, you can verify that a given output was produced by a specific version of the pipeline from a specific input. If a stakeholder questions a number, you can trace it back through the entire chain: output hash → pipeline version → input hash → raw data file. That traceability is what makes a pipeline trustworthy.

Automated Quality Checks

A pipeline without validation is a pipeline you do not trust. Automated quality checks run after every execution and verify that the clean data meets defined standards. If a check fails, the pipeline stops and alerts the team. Bad data does not propagate to reports and models.

Categories of Quality Checks

Completeness: Required columns have no nulls. Row counts are within expected ranges.

assert clean_df['customer_id'].notna().all(), "customer_id contains nulls"
assert clean_df['order_date'].notna().all(), "order_date contains nulls"
assert len(clean_df) > 10000, f"Row count {len(clean_df)} below threshold"

Uniqueness: Primary keys are unique. No duplicate rows.

assert clean_df['order_id'].is_unique, "order_id contains duplicates"

Validity: Values fall within allowed ranges and match expected formats.

assert (clean_df['revenue'] >= 0).all(), "negative revenue found"
assert (clean_df['order_date'] <= pd.Timestamp.today()).all(), "future date found"
assert clean_df['region'].isin(['NA', 'EU', 'APAC', 'LATAM']).all(), "invalid region value"

Consistency: Related columns agree. Foreign keys reference existing primary keys.

orphans = clean_df.merge(customers_df, on='customer_id', how='left', indicator=True)
assert (orphans['_merge'] == 'both').all(), "orphaned customer_id found"

Threshold-Based Alerts

Not every quality issue is a hard failure. Some are warnings that indicate potential problems:

null_pct = clean_df['optional_field'].isna().mean()
if null_pct > 0.10:
    log.warning(f"optional_field is {null_pct:.1%} null (threshold: 10%)")

Threshold-based alerts surface trends without blocking the pipeline. If the null rate in an optional field creeps from 5% to 15% over several runs, the alert reveals a problem before it becomes a crisis.

Monitoring and Trend Tracking

Track quality metrics over time. A sudden change in row count, null rate, or outlier percentage indicates an upstream problem.

metrics = {
    'run_date': pd.Timestamp.today(),
    'input_rows': len(raw_df),
    'output_rows': len(clean_df),
    'quarantined_rows': len(quarantine_df),
    'null_rate': clean_df.isna().mean().mean(),
    'outlier_count': clean_df['is_outlier'].sum()
}

metrics_df = pd.DataFrame([metrics])
metrics_df.to_csv('quality_metrics.csv', mode='a', header=False, index=False)

This metrics table becomes a time-series view of data quality. Plotting it reveals degradation over time, seasonal patterns, and the impact of pipeline changes.

The Human Review Gate

Automation catches known failure modes. It does not catch new, unforeseen problems. A human review gate — a person who reviews the cleaning log and quality metrics before the clean data is published — is the final line of defense.

The review does not need to examine every row. It examines the summary: How many rows were processed? How many were quarantined? Did any validation checks fail? Were there any new rejection reasons? If the answers are all normal, the data is published. If anything is abnormal, the reviewer investigates before releasing the data.

A pipeline with automated checks and a human review gate is trustworthy. A pipeline with neither is a liability.

Building a Validation Report

Automated checks produce pass/fail results. A validation report presents those results in a human-readable format that summarizes the health of the clean data.

def generate_validation_report(clean_df, quarantine_df, metrics):
    report = f"""
# Data Quality Report
## {metrics['run_date']}

### Summary
- Input rows: {metrics['input_rows']:,}
- Output rows: {metrics['output_rows']:,}
- Quarantined rows: {metrics['quarantined_rows']:,}
- Quarantine rate: {metrics['quarantined_rows']/metrics['input_rows']:.2%}

### Validation Results
- Completeness: {'PASS' if metrics['null_rate'] < 0.01 else 'FAIL'}
- Uniqueness: {'PASS' if metrics['duplicate_count'] == 0 else 'FAIL'}
- Validity: {'PASS' if metrics['invalid_count'] == 0 else 'FAIL'}
- Consistency: {'PASS' if metrics['orphan_count'] == 0 else 'FAIL'}

### Quarantine Breakdown
{quarantine_df['rejection_reason'].value_counts().to_string()}
"""
    return report

The report is written to a file and optionally emailed to the data team. It provides a snapshot of data quality at a glance: if all checks pass and the quarantine rate is normal, the data is ready for use. If any check fails or the quarantine rate spikes, the team investigates before releasing the data.

Continuous Improvement

A pipeline is not finished; it is maintained. Every production run is an opportunity to improve:

• New rejection reasons indicate new data problems that need new cleaning rules.
• Increasing quarantine rates indicate upstream degradation that needs to be reported to the data source team.
• New variants in categorical columns need new mapping entries.
• Performance degradation may indicate that the pipeline needs optimization.

Track these signals over time. A pipeline that is actively maintained gets better with every run. A pipeline that is set and forgotten degrades silently until it produces wrong results.