data-quality-framework

Lightweight, pluggable data-quality validation for pandas pipelines — schema checks, freshness, duplicates, referential integrity, and row-count SLAs, with HTML reporting and email alerting.

---

Overview / Aim

Silent data-quality regressions — a renamed upstream column, a stalled load, a duplicate-producing join, a broken foreign key — are among the most expensive failure modes in a data platform because they corrupt dashboards and models without throwing errors. This framework provides a small, dependency-light set of composable validators that run against any pandas.DataFrame and return structured results suitable for gating a pipeline, rendering a report, or firing an alert.

The design goal is expressiveness with near-zero ceremony: each check is a plain function or small dataclass-driven class, so it can be dropped into an Airflow task, a dbt post-hook wrapper, or a CI job without pulling in a heavy data-quality platform.

Architecture / How It Works

                 ┌──────────────────────────────────────────┐
   DataFrame ───▶│              Validators (src/validators)   │
                 │  schema · freshness · duplicates           │
                 │  referential · sla (row-count)             │
                 └───────────────────┬──────────────────────┘
                                     │  structured dict results
                                     ▼
                       ┌──────────────────────────┐
                       │  ValidationReport (report)│  ──▶  HTML report
                       └────────────┬─────────────┘
                                    │ on failure
                                    ▼
                         ┌────────────────────┐
                         │  send_alert (alerts)│  ──▶  SMTP email
                         └────────────────────┘

Each validator returns either a dict ({"ok": bool, ...}) or a per-column error map, so results compose cleanly. ValidationReport accumulates named checks and renders a status table to HTML; send_alert delivers failures over SMTP.

Tech Stack & Tools

Tool	Role
Python 3.10+	Implementation language
pandas	DataFrame inspection / validation primitives
dataclasses	Declarative ColumnSpec / SLASpec definitions
smtplib + email.mime	SMTP email alerting (TLS)
pytest	Unit tests for every validator

Project Structure

data-quality-framework/
├── src/
│   ├── validators/
│   │   ├── schema.py        # ColumnSpec + SchemaValidator: dtype/null/range/allowed-values
│   │   ├── freshness.py     # check_freshness: max-age guard on a timestamp column
│   │   ├── duplicates.py    # check_duplicates: dup count + percentage on a key subset
│   │   ├── referential.py   # check_referential: orphan FK detection (parent/child)
│   │   └── sla.py           # SLASpec + RowCountChecker: min/max rows + stale-count warning
│   ├── report.py            # ValidationReport: accumulate checks → HTML status table
│   └── alerts.py            # send_alert: SMTP/TLS email notification
├── tests/
│   ├── test_validators.py        # schema / SLA / freshness
│   └── test_extra_validators.py  # referential / duplicates
└── docs/
    └── runbook.md           # triage steps per failure class

Key Features / Highlights

• Declarative schema contracts — ColumnSpec(dtype, nullable, min_val, max_val, allowed) drives null checks, range bounds, and allowed-value sets, returning a per-column error map that pinpoints exactly what drifted.
• Freshness SLA — check_freshness computes the age of the newest record (timezone-aware) and flags data older than a configurable threshold.
• Duplicate detection — reports both an absolute duplicate count and a duplicate percentage over an arbitrary key subset.
• Referential integrity — surfaces orphan foreign keys plus a sample of the offending keys for fast debugging.
• Row-count SLAs — RowCountChecker guards min/max volume and warns when a count is identical to the prior run (a classic stale-load signal).
• Reporting + alerting — HTML status tables for humans, SMTP email for on-call escalation.
• Operational runbook — docs/runbook.md maps each failure class to concrete triage steps (upstream join, dedup DAG, renamed columns).

Challenges

• Timezone correctness in freshness checks — naive timestamps are coerced to UTC before age computation to avoid silently passing stale data.
• Actionable failure output — validators deliberately return samples of bad values / orphan keys rather than a bare boolean, so an on-call engineer can act without re-querying.
• Keeping the surface small — every check is intentionally a single function or tiny class to stay embeddable inside orchestration without a heavy framework.

Future Work

• Package as an installable library with a unified run_suite() entry point.
• Add a duplicates import fix and broaden type hints across modules.
• Pluggable sinks (Slack, PagerDuty, OpenLineage) alongside SMTP.
• Great Expectations / Soda interop adapters.
• Threshold-as-config (YAML) so non-engineers can tune SLAs.

Getting Started / Usage

pip install pandas pytest

import pandas as pd
from src.validators.schema import SchemaValidator, ColumnSpec
from src.validators.freshness import check_freshness
from src.report import ValidationReport

df = pd.read_parquet("orders.parquet")

spec = {"amount": ColumnSpec(dtype="float64", nullable=False, min_val=0)}
errors = SchemaValidator(spec).validate(df)

report = ValidationReport("orders")
report.add("schema", ok=not errors, detail=str(errors))
report.add("freshness", **{"ok": check_freshness(df, "created_at", 24)["ok"]})
html = report.to_html()

Run the test suite:

pytest tests/ -v

Email alerting reads SMTP settings from the environment (SMTP_HOST, SMTP_USER, SMTP_PASS).

Conclusion

Demonstrates pragmatic data-quality engineering: declarative schema contracts, freshness/volume SLAs, referential and duplicate checks, plus reporting and alerting — the building blocks of a trustworthy data platform, implemented cleanly and unit-tested.