Data Quality Testing & Validation â Frameworks, Automation & Best Practices
In this tutorial, you'll learn about Data Quality Testing & Validation. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Data quality testing validates data against defined expectations â completeness, uniqueness, freshness, validity, accuracy, and consistency â at every stage of the pipeline from ingestion to consumption.
What You'll Learn
By the end of this tutorial, you'll understand the six data quality dimensions, how to use Great Expectations for declarative validation, dbt tests for SQL-first quality checks, Soda for open-source monitoring, automated validation pipelines, and production alerting strategies.
Why It Matters
Bad data costs businesses 15-20% of revenue and destroys trust in analytics. Unlike code, data changes over time â what's valid today fails tomorrow without warning. Testing data is harder than testing code because data distributions shift, sources evolve schemas, and pipelines accumulate edge cases. DodaTech runs 500+ automated quality checks across all Doda Browser telemetry pipelines to ensure dashboard accuracy and alert within 60 seconds of a quality failure.
Real-World Use
Great Expectations is used at Airbnb across 10,000+ datasets for data drift detection. dbt tests validate every model transformation at JetBlue. Soda checks run on 1,000+ tables at HelloFresh to catch freshness and volume anomalies before they reach dashboards.
Data Quality Validation Architecture
flowchart LR
subgraph "Sources"
A[Raw Data] --> B[Bronze Layer]
end
subgraph "Validation Stages"
B --> C1[Schema Checks]
B --> C2[Freshness Checks]
B --> C3[Volume Checks]
end
subgraph "Framework"
C1 --> D1[Great Expectations]
C2 --> D2[Soda]
C3 --> D3[dbt Tests]
end
subgraph "Actions"
D1 --> E1{Pass?}
D2 --> E1
D3 --> E1
E1 -->|Yes| F[Load to Silver]
E1 -->|No| G[Quarantine + Alert]
end
style D1 fill:#f90,color:#fff
style D2 fill:#f90,color:#fff
style D3 fill:#f90,color:#fff
Prerequisites: Familiarity with Python and SQL. Understanding of data pipeline concepts. Experience with ETL or data warehousing helps.
The Six Data Quality Dimensions
Every quality test maps to one of these dimensions. Together they provide complete coverage:
| Dimension | What It Detects | Example |
|---|---|---|
| Completeness | Missing or null values | Column X must be 0% null |
| Uniqueness | Duplicate records | Primary key has no duplicates |
| Timeliness | Stale or delayed data | Data loaded within 1 hour of generation |
| Validity | Format or type violations | Email matches regex, age is integer |
| Accuracy | Correct values | Revenue sums match source system |
| Consistency | Cross-system agreement | Row count matches between tables |
# quality_dimensions.py
# Evaluate all six DQ dimensions on a dataset
import re
from datetime import datetime, timedelta
class DataQualityEvaluator:
def __init__(self, data, schema):
self.data = data
self.schema = schema
def completeness(self, column):
total = len(self.data)
non_null = sum(1 for row in self.data if row.get(column) is not None)
return round(non_null / total * 100, 2)
def uniqueness(self, column):
values = [row.get(column) for row in self.data if row.get(column) is not None]
total = len(values)
unique = len(set(values))
return round(unique / total * 100, 2) if total > 0 else 100.0
def validity(self, column, pattern):
total = len(self.data)
valid = sum(1 for row in self.data
if row.get(column) and re.match(pattern, str(row[column])))
return round(valid / total * 100, 2)
def timeliness(self, timestamp_column, max_age_hours=24):
now = datetime.now()
ages = []
for row in self.data:
ts = row.get(timestamp_column)
if ts:
age = (now - ts).total_seconds() / 3600
ages.append(age)
if not ages:
return 0.0
within_sla = sum(1 for a in ages if a <= max_age_hours)
return round(within_sla / len(ages) * 100, 2)
def evaluate_all(self):
results = {}
for col, rules in self.schema.items():
col_results = {}
if "completeness" in rules:
col_results["completeness"] = self.completeness(col)
if "uniqueness" in rules:
col_results["uniqueness"] = self.uniqueness(col)
if "validity" in rules:
col_results["validity"] = self.validity(col, rules["validity"])
results[col] = col_results
for rule in self.schema.get("_global", []):
if rule["type"] == "timeliness":
results["_timeliness"] = self.timeliness(rule["column"], rule.get("max_hours", 24))
return results
sample_data = [
{"email": "alice"@example".com", "order_id": "ORD-001", "created_at": datetime.now() - timedelta(hours=2)},
{"email": "bob"@example".com", "order_id": "ORD-002", "created_at": datetime.now() - timedelta(hours=4)},
{"email": None, "order_id": "ORD-003", "created_at": datetime.now() - timedelta(hours=6)},
{"email": "bad-email", "order_id": "ORD-001", "created_at": datetime.now() - timedelta(hours=48)},
]
schema = {
"email": {"completeness": True, "validity": r"^[\w.+-]+@[\w-]+\.[\w.]+$"},
"order_id": {"completeness": True, "uniqueness": True},
"_global": [{"type": "timeliness", "column": "created_at", "max_hours": 24}],
}
evaluator = DataQualityEvaluator(sample_data, schema)
results = evaluator.evaluate_all()
for col, dims in results.items():
print(f"{col}: {dims}")
Expected output:
email: {'completeness': 75.0, 'validity': 50.0}
order_id: {'completeness': 100.0, 'uniqueness': 66.67}
_timeliness: 75.0
Great Expectations for Declarative Validation
Great Expectations (GE) lets you define expectations about your data â assertions similar to unit tests but for datasets.
# great_expectations_demo.py
# Simulate Great Expectations suite
class ExpectationsSuite:
def __init__(self, suite_name):
self.name = suite_name
self.expectations = []
self.results = []
def expect_column_values_to_not_be_null(self, column):
self.expectations.append(("not_null", {"column": column}))
def expect_column_values_to_be_unique(self, column):
self.expectations.append(("unique", {"column": column}))
def expect_column_values_to_match_regex(self, column, regex):
self.expectations.append(("match_regex", {"column": column, "regex": regex}))
def expect_column_pair_values_to_be_equal(self, col_a, col_b):
self.expectations.append(("pair_equal", {"col_a": col_a, "col_b": col_b}))
def run(self, data):
self.results = []
for exp_type, kwargs in self.expectations:
if exp_type == "not_null":
col = kwargs["column"]
passed = all(row.get(col) is not None for row in data)
self.results.append({
"expectation": f"expect_column_values_to_not_be_null('{col}')",
"passed": passed,
"observed": f"{sum(1 for r in data if r.get(col) is not None)}/{len(data)} non-null",
})
elif exp_type == "unique":
col = kwargs["column"]
values = [r.get(col) for r in data if r.get(col) is not None]
passed = len(values) == len(set(values))
self.results.append({
"expectation": f"expect_column_values_to_be_unique('{col}')",
"passed": passed,
"observed": f"{len(values)} values, {len(set(values))} unique",
})
elif exp_type == "match_regex":
col = kwargs["column"]
regex = kwargs["regex"]
import re
passed = all(re.match(regex, str(r.get(col, "")))
for r in data if r.get(col) is not None)
self.results.append({
"expectation": f"expect_column_values_to_match_regex('{col}')",
"passed": passed,
"observed": f"pattern={'yes' if passed else 'no'}",
})
elif exp_type == "pair_equal":
col_a, col_b = kwargs["col_a"], kwargs["col_b"]
passed = all(r.get(col_a) == r.get(col_b) for r in data)
self.results.append({
"expectation": f"expect_column_pair_values_to_be_equal('{col_a}', '{col_b}')",
"passed": passed,
"observed": f"{'all equal' if passed else 'mismatch found'}",
})
return self.results
def summary(self):
total = len(self.results)
passed_count = sum(1 for r in self.results if r["passed"])
print(f"\n=== Great Expectations: {self.name} ===")
for r in self.results:
status = "PASS" if r["passed"] else "FAIL"
print(f" [{status}] {r['expectation']} -> {r['observed']}")
print(f"\nPassed: {passed_count}/{total}")
suite = ExpectationsSuite("orders_quality")
suite.expect_column_values_to_not_be_null("order_id")
suite.expect_column_values_to_be_unique("order_id")
suite.expect_column_values_to_match_regex("order_id", r"^ORD-\d{3}$")
suite.expect_column_values_to_not_be_null("customer_email")
suite.expect_column_values_to_match_regex("customer_email", r"^[\w.+-]+@[\w-]+\.[\w.]+$")
order_data = [
{"order_id": "ORD-001", "customer_email": "alice"@example".com"},
{"order_id": "ORD-002", "customer_email": "bob"@example".com"},
{"order_id": "ORD-003", "customer_email": None},
{"order_id": "ORD-001", "customer_email": "charlie"@example".com"},
{"order_id": "INVALID", "customer_email": "not-an-email"},
]
suite.run(order_data)
suite.summary()
Expected output:
=== Great Expectations: orders_quality ===
[PASS] expect_column_values_to_not_be_null('order_id') -> 5/5 non-null
[FAIL] expect_column_values_to_be_unique('order_id') -> 5 values, 4 unique
[FAIL] expect_column_values_to_match_regex('order_id') -> pattern=no
[FAIL] expect_column_values_to_not_be_null('customer_email') -> 4/5 non-null
[FAIL] expect_column_values_to_match_regex('customer_email') -> pattern=no
Passed: 1/5
Soda for Open-Source Monitoring
Soda is an open-source data quality framework that scans tables and runs checks using YAML configuration.
# soda_checks.yml
checks for orders_table:
- row_count > 0
- freshness(created_at) < 24h
- missing_count(order_id) = 0
- duplicate_count(order_id) = 0
- invalid_count(email) = 0:
valid_format: email
- avg_duration between 100 and 50000
- max(amount) < 100000
- min(amount) >= 0
# soda_scanner.py
# Simulate Soda scan execution
class SodaScanner:
def __init__(self, dataset_name):
self.name = dataset_name
self.checks = []
self.results = []
def add_check(self, check_type, params):
self.checks.append({"type": check_type, "params": params})
def scan(self, data):
self.results = []
for check in self.checks:
result = self._execute_check(check, data)
self.results.append(result)
return self.results
def _execute_check(self, check, data):
check_type = check["type"]
params = check["params"]
if check_type == "row_count":
threshold = params.get("min", 1)
passed = len(data) >= threshold
return {"check": f"row_count >= {threshold}", "passed": passed,
"observed": len(data)}
elif check_type == "freshness":
column = params["column"]
max_hours = params.get("max_hours", 24)
from datetime import datetime
timestamps = [row.get(column) for row in data if row.get(column)]
if timestamps:
max_age = (datetime.now() - max(timestamps)).total_seconds() / 3600
else:
max_age = float("inf")
passed = max_age <= max_hours
return {"check": f"freshness({column}) < {max_hours}h", "passed": passed,
"observed": f"{max_age:.1f}h"}
elif check_type == "missing_count":
column = params["column"]
missing = sum(1 for row in data if row.get(column) is None)
passed = missing == 0
return {"check": f"missing_count({column}) = 0", "passed": passed,
"observed": missing}
elif check_type == "duplicate_count":
column = params["column"]
values = [row.get(column) for row in data if row.get(column) is not None]
duplicates = len(values) - len(set(values))
passed = duplicates == 0
return {"check": f"duplicate_count({column}) = 0", "passed": passed,
"observed": duplicates}
elif check_type == "avg_between":
column = params["column"]
values = [row.get(column) for row in data if row.get(column) is not None]
avg = sum(values) / len(values) if values else 0
passed = params.get("min", 0) <= avg <= params.get("max", float("inf"))
return {"check": f"avg({column}) between [{params.get('min')}, {params.get('max')}]",
"passed": passed, "observed": round(avg, 2)}
return {"check": "unknown", "passed": False, "observed": "N/A"}
def report(self):
total = len(self.results)
passed = sum(1 for r in self.results if r["passed"])
print(f"\n=== Soda Scan: {self.name} ===")
for r in self.results:
status = "PASS" if r["passed"] else "FAIL"
print(f" [{status}] {r['check']} -> observed: {r['observed']}")
print(f"\nPassed: {passed}/{total}")
scanner = SodaScanner("orders_table")
scanner.add_check("row_count", {"min": 1})
scanner.add_check("freshness", {"column": "created_at", "max_hours": 24})
scanner.add_check("missing_count", {"column": "order_id"})
scanner.add_check("duplicate_count", {"column": "order_id"})
scanner.add_check("avg_between", {"column": "amount", "min": 10, "max": 500})
from datetime import datetime, timedelta
scan_data = [
{"order_id": "ORD-001", "amount": 150, "created_at": datetime.now() - timedelta(hours=2)},
{"order_id": "ORD-002", "amount": 250, "created_at": datetime.now() - timedelta(hours=5)},
{"order_id": None, "amount": 25, "created_at": datetime.now() - timedelta(hours=48)},
{"order_id": "ORD-001", "amount": 300, "created_at": datetime.now() - timedelta(hours=1)},
]
scanner.scan(scan_data)
scanner.report()
Expected output:
=== Soda Scan: orders_table ===
[PASS] row_count >= 1 -> observed: 4
[FAIL] freshness(created_at) < 24h -> observed: 48.0h
[FAIL] missing_count(order_id) = 0 -> observed: 1
[FAIL] duplicate_count(order_id) = 0 -> observed: 1
[PASS] avg(amount) between [10, 500] -> observed: 181.25
Passed: 2/5
Automated Validation Pipeline
Combine all tools into a single automated pipeline that runs on every data load:
# validation_pipeline.py
class ValidationPipeline:
def __init__(self, name):
self.name = name
self.steps = []
self.scores = {}
def add_step(self, step_name, validation_fn, weight=1.0):
self.steps.append({"name": step_name, "fn": validation_fn, "weight": weight})
def run(self, data):
print(f"\n{'='*55}")
print(f" Validation Pipeline: {self.name}")
print(f"{'='*55}")
for step in self.steps:
print(f"\n Step: {step['name']}")
result = step["fn"](data)
self.scores[step["name"]] = result
for k, v in result.items():
print(f" {k}: {v}")
weighted_sum = sum(
self.scores[s["name"]].get("score", 0) * s["weight"]
for s in self.steps
)
total_weight = sum(s["weight"] for s in self.steps)
overall = round(weighted_sum / total_weight, 1) if total_weight > 0 else 0
print(f"\n{'='*55}")
print(f" Overall Quality Score: {overall}%")
if overall >= 95:
print(f" Status: PASS - Loading data")
elif overall >= 80:
print(f" Status: WARN - Loading with alert")
else:
print(f" Status: FAIL - Blocking pipeline")
print(f"{'='*55}")
return overall
def schema_check(data):
expected = {"order_id", "amount", "created_at"}
actual = set(data[0].keys()) if data else set()
missing = expected - actual
score = 100 if not missing else 50
return {"expected_columns": list(expected), "missing_columns": list(missing), "score": score}
def completeness_check(data, column="order_id"):
total = len(data)
nulls = sum(1 for r in data if not r.get(column))
score = round((total - nulls) / total * 100, 1)
return {"total": total, f"null_{column}": nulls, "score": score}
def freshness_check(data, column="created_at", max_hours=24):
from datetime import datetime
timestamps = [r.get(column) for r in data if r.get(column)]
if timestamps:
max_age = (datetime.now() - max(timestamps)).total_seconds() / 3600
else:
max_age = float("inf")
score = 100 if max_age <= max_hours else 0
return {"max_age_hours": round(max_age, 1), "threshold_hours": max_hours, "score": score}
pipeline = ValidationPipeline("Clickstream Quality")
pipeline.add_step("Schema Validation", schema_check, weight=1.0)
pipeline.add_step("Completeness Check", lambda d: completeness_check(d, "order_id"), weight=1.0)
pipeline.add_step("Freshness Check", freshness_check, weight=2.0)
from datetime import datetime, timedelta
sample = [
{"order_id": "ORD-001", "amount": 100, "created_at": datetime.now() - timedelta(hours=1)},
{"order_id": "ORD-002", "amount": 200, "created_at": datetime.now() - timedelta(hours=3)},
{"order_id": None, "amount": 50, "created_at": datetime.now() - timedelta(hours=48)},
]
pipeline.run(sample)
Expected output:
=======================================================
Validation Pipeline: Clickstream Quality
=======================================================
Step: Schema Validation
expected_columns: ['amount', 'order_id', 'created_at']
missing_columns: []
score: 100
Step: Completeness Check
total: 3
null_order_id: 1
score: 66.7
Step: Freshness Check
max_age_hours: 48.0
threshold_hours: 24
score: 0
=======================================================
Overall Quality Score: 41.7%
Status: FAIL - Blocking pipeline
=======================================================
Common Data Quality Testing Mistakes
1. Testing Only at the End
Running quality checks only on the final table hides problems deep in the pipeline. Check at every layer (bronze, silver, gold) so failures surface immediately.
2. No Freshness Alerts
A pipeline that loads stale data silently is worse than a broken pipeline â dashboards look normal but show wrong answers. Always set freshness SLAs with alerts.
3. Tests That Never Fail
If all your tests always pass, they're testing the wrong things. Review test coverage quarterly. Add edge case expectations: nulls in non-nullable columns, future dates, negative values.
4. Ignoring Data Distribution Drift
Schema tests catch structural changes, but distribution drift â like "status" suddenly having 50% nulls instead of 2% â goes undetected. Monitor distributions with profiling.
5. Manual Quality Checks
Engineers running ad-hoc SQL queries for quality checks is unsustainable at scale. Automate every check, route failures to a central dashboard, and alert on thresholds.
Practice Questions
1. What are the six data quality dimensions and how do they differ? Completeness (no nulls), uniqueness (no duplicates), timeliness (fresh data), validity (correct format), accuracy (values match reality), consistency (agreement across systems). Each detects a different class of data problem; all six are needed for comprehensive coverage.
2. How does Great Expectations differ from dbt tests? Great Expectations is declarative and engine-agnostic â expectations are defined in Python and can run against any data source. dbt tests are SQL-based, run inside the warehouse, and integrate natively with dbt models. Many teams use both: GE for raw data validation, dbt tests for transformed models.
3. What is data profiling and when would you use it? Profiling analyzes column statistics: null percentage, unique count, min/max, distribution shape, frequent values. Use it to discover data characteristics before writing tests, to suggest expectations automatically, and to detect distribution drift over time.
Frequently Asked Questions
{{< faq question="What is the difference between data quality testing and data monitoring?">}} Testing runs on demand or on schedule to validate data at a point in time (e.g., after a dbt run). Monitoring continuously tracks metrics over time â row counts, freshness, distribution â and alerts on anomalies. Testing is proactive validation; monitoring is reactive detection. Both are required for production data quality. {{< /faq >}}
{{< faq question="How many tests should I have per table?">}} Start with 5-10 tests per table: not_null on primary and foreign keys, unique on primary key, accepted_values on categorical columns, freshness on time columns, row_count thresholds. Add business-specific tests as you discover edge cases. A good signal: if all tests pass for 30 days straight, add harder tests. If tests break frequently, they're catching real problems. {{< /faq >}}
Mini Project: Quality Score Tracker
# quality_tracker.py
# Track data quality scores over time
import random
from datetime import datetime, timedelta
class QualityTracker:
def __init__(self):
self.scores = {}
self.alerts = []
def record(self, pipeline, score):
if pipeline not in self.scores:
self.scores[pipeline] = []
self.scores[pipeline].append({"timestamp": datetime.now(), "score": score})
if score < 80:
self.alerts.append({
"pipeline": pipeline,
"timestamp": datetime.now(),
"score": score,
"severity": "CRITICAL" if score < 60 else "WARNING",
})
def summary(self):
print(f"\n{'='*55}")
print(f" Data Quality Dashboard")
print(f"{'='*55}")
for pipeline, entries in sorted(self.scores.items()):
latest = entries[-1]["score"]
avg = sum(e["score"] for e in entries) / len(entries)
bar = "#" * int(latest // 5)
trend = "UP" if len(entries) > 1 and entries[-1]["score"] > entries[-2]["score"] else "DOWN"
print(f"\n {pipeline}")
print(f" Latest: {latest}% ({trend}) | Avg: {avg:.1f}%")
print(f" {bar}")
print(f"\n Active Alerts: {len(self.alerts)}")
for a in self.alerts[-3:]:
print(f" [{a['severity']}] {a['pipeline']}: {a['score']}%")
tracker = QualityTracker()
tracker.record("clickstream", 95)
tracker.record("clickstream", 92)
tracker.record("orders", 88)
tracker.record("clickstream", 73)
tracker.record("orders", 96)
tracker.record("ml_features", 45)
tracker.summary()
Expected output:
=======================================================
Data Quality Dashboard
=======================================================
clickstream
Latest: 73% (DOWN) | Avg: 86.7%
###############
orders
Latest: 96% (UP) | Avg: 92.0%
###################
ml_features
Latest: 45% (DOWN) | Avg: 45.0%
#########
Active Alerts: 2
[WARNING] clickstream: 73%
[CRITICAL] ml_features: 45%
Related Concepts
What's Next
You now understand data quality testing frameworks and how to automate validation in production. Next, learn how Apache Airflow orchestrates these checks in pipeline DAGs, and explore Cloud Computing services for managed quality monitoring.
- Practice daily â Add Great Expectations or dbt tests to one existing table this week
- Build a project â Create a validation pipeline that checks schema, freshness, and volume on every data load
- Explore related topics â Check out data Contract Testing, schema registries, and data Observability platforms
Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro