Data is often called the “new oil,” but anyone who works with it knows that most of it is crude. It’s messy, inconsistent, and sometimes plain wrong.
If you feed raw data directly into analytics dashboards or machine learning models, you get unreliable results. The traditional approach—writing ad-hoc SQL queries to check for nulls—doesn’t scale.
We need to stop treating Data Quality (DQ) as an afterthought and start treating it as an engineering discipline.
In this article, I will walk through building a production-ready Data Quality Pipeline. We will move beyond simple validation and implement an “Auto-Healing” system that fixes common errors automatically before quarantining the truly bad data.
We will build this using the industry-standard stack for robust data orchestration: Python, Apache Airflow, and Docker.
The Philosophy: A “Bronze to Silver” Strategy
We are adopting a “Medallion Architecture” approach. Our goal is to take Bronze data (raw, messy files landing in our data lake) and refine it into Silver data (clean, conformed, and validated tables ready for analysis).
To achieve this, our pipeline rests on three pillars:
1. Auto-Healing (Sanitization)
Before we complain about bad data, we should try to fix it. If a user enters their email as ” Bob@Example.COM “, we know what they meant. Rejecting this record creates unnecessary noise.
Our Approach: We will build a Python “Healer” engine that trims whitespace, standardizes casing, and formats messy inputs like UK Postcodes before validation occurs.
2. Rule-Based Validation (The Contract)
Once data is sanitized, we enforce strict business rules.
Our Approach: We decouple the rules from the code. Instead of hardcoding if value < 0: raise Error, we define rules in a JSON configuration file. This allows us to apply regex checks or complex custom SQL logic dynamically.
3. Quarantine, Don’t Crash
A common mistake is crashing the entire pipeline when one bad row is detected. This stops the flow of good data.
Our Approach: We implement the “Audit Pattern.” Bad rows are captured, tagged with an error message, and written to a persistent error_out database table. The pipeline finishes successfully, and analysts can review the quarantine table later.
The Tech Stack: Why Airflow and Docker?
To build a system this robust, we need the right tools.
- Python (Pandas/SQLite): The engine for data manipulation and persistent storage of our audit trail.
- Apache Airflow: The orchestrator. It manages the dependencies between tasks (ensuring we don’t validate before we heal), handles retries, and provides observability into failures.
- Docker: The equalizer. It packages Airflow, Python, and our database into a single, consistent environment. It ensures the code runs exactly the same on my laptop as it does in production.
The Architecture
Our project adheres to a strict separation of concerns. We keep our orchestration logic separate from our business logic.
Directory Structure
Plaintext
my_dq_project/
├── docker-compose.yml # The environment blueprint
├── Dockerfile # The container definition
├── dags/
│ └── dq_csv_dag.py # The Airflow workflow definition
├── src/
│ ├── data_healer.py # The Auto-Healing logic class
│ └── dq_engine.py # The Validation & Database logic class
└── data/
├── incoming_data.csv # The input file
└── dq_database.db # The local SQLite output DB
dags/: Contains only Airflow code. These files define when things run and in what order.src/: Contains pure Python classes. These files define how data is processed. They know nothing about Airflow.data/: A shared volume for inputs and outputs.
Step 1: The Auto-Healing Engine (src/)
The core of our strategy is the DataHealer class. It uses a “Strategy Pattern.” We define atomic transformation functions (like trimming space) and then configure which columns get which strategies in a JSON config.
Here is a snippet of the healing logic for a complex case, like standardizing UK Postcodes:
Python
# src/data_healer.py snippet
def _format_uk_postcode(self, val):
"""
Heuristic: Standardizes messy inputs like 'sw1a1aa'
into the official format 'SW1A 1AA'.
"""
# Normalize: Remove spaces and uppercase
s_val = str(val).replace(" ", "").upper()
# Safety Check: If it's too short or long to be a postcode,
# return as-is and let the validator catch it later.
if len(s_val) < 5 or len(s_val) > 7:
return s_val
# Slice and rejoin with a single space
return f"{s_val[:-3]} {s_val[-3:]}"
By configuring our pipeline to apply this strategy to postcode columns, we fix thousands of minor data entry errors without human intervention.
Step 2: The Dynamic Validation Engine
Once healed, the data must pass our “Quality Contract.” This is defined in a simple JSON array. This approach makes the system incredibly flexible.
JSON
[
{
"column": "contact_email",
"function": "check_custom_sql",
"params": {"sql_expression": "{col} LIKE '%_@__%.__%'"},
"error_msg": "Invalid email format"
},
{
"column": "billing_code",
"function": "check_uk_postcode",
"error_msg": "Invalid UK postcode format"
}
]
Our Python DataQualityEngine reads this JSON and dynamically applies the checks. Rows that fail are not discarded; they are structured and written to our SQLite audit table:
| run_id | row_id | column | invalid_value | error_message |
| batch_001 | 501 | billing_code | 90210 | Invalid UK postcode format |
| batch_001 | 502 | contact_email | bob#gmail.com | Invalid email format |
Step 3: Orchestration with Airflow DAGs
We use Airflow to stitch these Python classes together into a reliable workflow.
Our DAG (Directed Acyclic Graph) defines a clear, linear dependency chain. We use Airflow’s XComs (Cross-Communication) to pass data between the tasks.
- Extract Task: Reads the CSV file from disk and pushes raw data to XCom.
- Heal Task: Pulls raw data from XCom, runs the
DataHealer, pushes clean data to XCom. - Validate Task: Pulls clean data, runs the
DataQualityEngine, and saves errors to the database.
The Power of Airflow Variables
Hardcoding filenames in code is bad practice. We leveraged Airflow Variables to decouple configuration.
Instead of hardcoding incoming_data.csv, our DAG asks Airflow for the filename:
Python
# dags/dq_csv_dag.py snippet
from airflow.models import Variable
def load_csv_data(**kwargs):
# Fetch filename from UI, default to 'incoming_data.csv'
filename = Variable.get("target_csv_file", default_var="incoming_data.csv")
file_path = f'/opt/airflow/data/{filename}'
# ... execute load ...
This allows an Operations person to change the input file target via the Airflow Web UI without ever touching the Python code.
Step 4: Running it with Docker
The final piece of the puzzle is the environment. We use Docker Compose to spin up the entire stack—Airflow Webserver, Scheduler, and our Python environment—with one command.
The crucial part of our docker-compose.yml is Volume Mounting:
YAML
volumes:
# Mount our local source code into the container's python path
- ./src:/opt/airflow/src
# Mount our local data folder so the SQLite DB persists on our laptop
- ./data:/opt/airflow/data
This setup means we can edit Python code in our favorite local editor (like VS Code), and the running Airflow container sees the changes instantly. Furthermore, the SQLite database created inside the container is saved to our local hard drive, so the data persists even if the container is destroyed.
To start the entire platform, we simply run:
Bash
docker-compose up
We can then access the Airflow UI at http://localhost:8080, trigger our pipeline, and watch the data flow from raw CSV to healed, validated insights.
Conclusion
By moving from ad-hoc scripts to an engineered pipeline using Airflow and Docker, we gain immense benefits:
- Reproducibility: The environment is identical in development and production.
- Observability: Airflow provides visual logs and status for every step.
- Resilience: Auto-healing reduces noise, and the quarantine pattern ensures we never lose data.
Data quality is not a one-time project; it’s a continuous process. Building a robust foundation like this ensures your data team spends less time fixing broken pipelines and more time delivering value.