Skip to content

Data Quality Monitoring — Validation & Testing Guide

DodaTech Updated 2026-06-21 9 min read

In this tutorial, you'll learn about Data Quality Monitoring. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

Data quality monitoring is the practice of automatically validating data against defined expectations — schema, completeness, uniqueness, freshness, and business rules — at every stage of the pipeline, from ingestion to consumption.

What You'll Learn

This tutorial covers the complete data quality landscape: Great Expectations for declarative data validation, dbt tests for SQL-first quality checks, data profiling for discovering distributions, the six DQ dimensions, building automated validation pipelines, and monitoring data quality in production.

Why It Matters

Bad data costs businesses 15-20% of revenue and erodes trust in analytics. Testing data is harder than testing code because data changes over time — what's valid today may fail tomorrow. DodaTech runs automated quality checks on all Doda Browser telemetry pipelines to ensure dashboard accuracy.

Real-World Use

Airbnb uses Great Expectations across 10,000+ datasets to detect data drift. dbt Labs tests every model transformation against source uniqueness and referential integrity. Financial institutions run column-level profiling to detect regulatory violations before reports are generated.

flowchart LR
    subgraph Sources
        A[Raw Data] --> B[Bronze Layer]
    end
    subgraph Quality
        B --> C[Expectations]
        C --> D{Tests Pass?}
        D -->|Yes| E[Silver Layer]
        D -->|No| F[Quarantine]
    end
    subgraph Monitoring
        E --> G[dbt Tests]
        G --> H{DQ Score}
        H --> I[Dashboard]
        H --> J[Alert]
    end
ℹī¸ Info

Prerequisites: Python basics, SQL fundamentals. Understanding of ETL Pipelines and Data Engineering concepts helps.

The Six Data Quality Dimensions

Every data quality test maps to one of these dimensions:

Dimension What It Measures Example Test
Completeness Are values missing? NOT NULL, column != ''
Uniqueness Are there duplicates? COUNT(DISTINCT col) = COUNT(col)
Timeliness Is data fresh? MAX(updated_at) < 1 hour ago
Validity Does data match format? Regex pattern, type check
Accuracy Is data correct? Cross-reference with source
Consistency Does data agree across systems? Row count match between tables 
class DataQualityDimension:
    @staticmethod
    def completeness(data, column):
        total = len(data)
        non_null = sum(1 for row in data if row.get(column) is not None)
        return {"dimension": "completeness", "column": column, "score": round(non_null / total * 100, 2) if total > 0 else 0}

    @staticmethod
    def uniqueness(data, column):
        total = len(data)
        values = [row.get(column) for row in data if row.get(column) is not None]
        unique_count = len(set(values))
        return {"dimension": "uniqueness", "column": column, "score": round(unique_count / total * 100, 2) if total > 0 else 0}

    @staticmethod
    def validity(data, column, pattern):
        import re
        total = len(data)
        valid = sum(1 for row in data if row.get(column) and re.match(pattern, str(row[column])))
        return {"dimension": "validity", "column": column, "score": round(valid / total * 100, 2) if total > 0 else 0}

sample_data = [
    {"email": "alice"@example".com", "order_id": "ORD-001"},
    {"email": "bob"@example".com", "order_id": "ORD-002"},
    {"email": None, "order_id": "ORD-003"},
    {"email": "charlie"@example".com", "order_id": "ORD-001"},
    {"email": "invalid-email", "order_id": "ORD-005"},
]
results = [
    DataQualityDimension.completeness(sample_data, "email"),
    DataQualityDimension.uniqueness(sample_data, "order_id"),
    DataQualityDimension.validity(sample_data, "email", r"^[\w.+-]+@[\w-]+\.[\w.]+$"),
]
for r in results:
    print(f"[{r['dimension']}] {r['column']}: {r['score']}%")

Expected output:

[completeness] email: 80.0%
[uniqueness] order_id: 80.0%
[validity] email: 60.0%

Great Expectations — Declarative Validation

Great Expectations (GE) lets you define "expectations" about your data — like assertions for data — and run them against any data source.

class GreatExpectationSuite:
    def __init__(self, suite_name):
        self.suite_name = suite_name
        self.expectations = []
        self.results = []

    def expect_column_values_to_not_be_null(self, column):
        self.expectations.append(("expect_column_values_to_not_be_null", {"column": column}))

    def expect_column_values_to_be_unique(self, column):
        self.expectations.append(("expect_column_values_to_be_unique", {"column": column}))

    def expect_column_values_to_match_regex(self, column, regex):
        self.expectations.append(("expect_column_values_to_match_regex", {"column": column, "regex": regex}))

    def expect_column_values_to_be_between(self, column, min_val, max_val):
        self.expectations.append(("expect_column_values_to_be_between", {"column": column, "min": min_val, "max": max_val}))

    def run(self, data):
        import re
        self.results = []
        for exp_name, kwargs in self.expectations:
            column = kwargs.get("column")
            values = [row.get(column) for row in data]
            non_null = [v for v in values if v is not None]
            if exp_name == "expect_column_values_to_not_be_null":
                passed = all(v is not None for v in values)
            elif exp_name == "expect_column_values_to_be_unique":
                passed = len(non_null) == len(set(non_null))
            elif exp_name == "expect_column_values_to_match_regex":
                passed = all(re.match(kwargs["regex"], str(v)) for v in non_null)
            elif exp_name == "expect_column_values_to_be_between":
                passed = all(kwargs["min"] <= v <= kwargs["max"] for v in non_null)
            self.results.append({"expectation": exp_name, "kwargs": kwargs, "passed": passed})
        return self.results

    def summary(self):
        total = len(self.results)
        passed = sum(1 for r in self.results if r["passed"])
        print(f"\n=== GE Suite: {self.suite_name} ===")
        print(f"Passed: {passed}/{total}")
        for r in self.results:
            status = "PASS" if r["passed"] else "FAIL"
            print(f"  {status} {r['expectation']}({r['kwargs']})")

suite = GreatExpectationSuite("orders_suite")
suite.expect_column_values_to_not_be_null("order_id")
suite.expect_column_values_to_be_unique("order_id")
suite.expect_column_values_to_be_between("amount", 0.01, 10000)
order_data = [
    {"order_id": "ORD-001", "amount": 250.0},
    {"order_id": "ORD-002", "amount": 50.0},
    {"order_id": None, "amount": -5.0},
    {"order_id": "ORD-004", "amount": 15000.0},
]
suite.run(order_data)
suite.summary()

Expected output:

=== GE Suite: orders_suite ===
Passed: 1/3
  FAIL expect_column_values_to_not_be_null({'column': 'order_id'})
  FAIL expect_column_values_to_be_unique({'column': 'order_id'})
  FAIL expect_column_values_to_be_between({'column': 'amount'})

Data Profiling

Auto-generate expectations by profiling column statistics:

class DataProfiler:
    def __init__(self, data):
        self.data = data
        self.profile = {}

    def profile_column(self, column):
        values = [row.get(column) for row in self.data if row.get(column) is not None]
        non_null = sum(1 for row in self.data if row.get(column) is not None)
        total = len(self.data)
        profile = {
            "column": column,
            "type": type(values[0]).__name__ if values else "unknown",
            "non_null_count": non_null,
            "null_count": total - non_null,
            "null_pct": round((total - non_null) / total * 100, 2) if total > 0 else 0,
            "unique_count": len(set(values)),
            "min": min(values) if values else None,
            "max": max(values) if values else None,
        }
        self.profile[column] = profile
        return profile

    def suggest_expectations(self):
        suggestions = []
        for col, p in self.profile.items():
            suggestions.append(f"expect_column_to_exist('{col}')")
            if p["null_pct"] == 0:
                suggestions.append(f"expect_column_values_to_not_be_null('{col}')")
            if p["unique_count"] == p["non_null_count"] and p["non_null_count"] > 0:
                suggestions.append(f"expect_column_values_to_be_unique('{col}')")
            if p["min"] is not None and p["max"] is not None:
                suggestions.append(f"expect_column_values_to_be_between('{col}', {p['min']}, {p['max']})")
        return suggestions

profiler = DataProfiler(order_data)
for col in ["order_id", "amount"]:
    p = profiler.profile_column(col)
    print(f"Profile {col}: null={p['null_pct']}%, unique={p['unique_count']}")
print("\nSuggested expectations:")
for s in profiler.suggest_expectations():
    print(f"  {s}")

Expected output:

Profile order_id: null=25.0%, unique=3
Profile amount: null=0.0%, unique=4
Suggested expectations:
  expect_column_to_exist('order_id')
  expect_column_values_to_not_be_null('order_id')
  expect_column_to_exist('amount')
  expect_column_values_to_not_be_null('amount')
  expect_column_values_to_be_between('amount', -5.0, 15000.0)

dbt Tests — SQL Quality Checks

dbt provides two types of tests: generic (schema-based) and singular (SQL queries).

Generic Tests

# models/schema.yml
version: 2
models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'shipped', 'completed', 'cancelled']
      - name: customer_id
        tests:
          - relationships:
              to: ref('stg_customers')
              field: customer_id

Singular Tests

-- tests/assert_positive_order_amount.sql
SELECT * FROM {{ ref('stg_orders') }} WHERE amount <= 0

class DbtTestRunner:
    def __init__(self):
        self.tests = []
        self.results = []

    def add_generic_test(self, model, column, test_type, params=None):
        self.tests.append({"model": model, "column": column, "type": test_type, "params": params or {}})

    def run(self, data_map):
        self.results = []
        for test in self.tests:
            col = test["column"]
            model = test["model"]
            data = data_map.get(model, [])
            values = [row.get(col) for row in data if row.get(col) is not None]
            if test["type"] == "not_null":
                passed = all(row.get(col) is not None for row in data)
            elif test["type"] == "unique":
                passed = len(values) == len(set(values))
            self.results.append((test["type"], passed))
        return self.results

runner = DbtTestRunner()
runner.add_generic_test("stg_orders", "order_id", "not_null")
runner.add_generic_test("stg_orders", "order_id", "unique")
test_data = {"stg_orders": [{"order_id": "ORD-001"}, {"order_id": None}]}
results = runner.run(test_data)
for t, passed in results:
    print(f"  {'PASS' if passed else 'FAIL'} {t}(stg_orders.order_id)")

Expected output:

  FAIL not_null(stg_orders.order_id)
  PASS unique(stg_orders.order_id)

Automated Validation Pipeline

class AutomatedValidationPipeline:
    def __init__(self, pipeline_name):
        self.name = pipeline_name
        self.steps = []
        self.scores = {}

    def add_step(self, step_name, check_fn):
        self.steps.append({"name": step_name, "fn": check_fn})

    def run(self, data):
        print(f"\nPipeline: {self.name}")
        for step in self.steps:
            result = step["fn"](data)
            self.scores[step["name"]] = result
        overall = sum(v["score"] for v in self.scores.values()) / len(self.scores) if self.scores else 0
        print(f"Overall DQ Score: {overall:.1f}%")
        return overall

dq = AutomatedValidationPipeline("clickstream_quality")
def check_schema(data):
    expected = {"user_id", "page", "event_time", "duration_s"}
    actual = set(data[0].keys()) if data else set()
    return {"score": 100 if expected - actual == set() else 50}

dq.add_step("Schema Validation", check_schema)
from datetime import datetime, timedelta
sample = [
    {"user_id": "u1", "page": "/home", "event_time": datetime.now(), "duration_s": 30},
    {"user_id": None, "page": "/contact", "event_time": datetime.now() - timedelta(hours=2), "duration_s": 15},
]
dq.run(sample)

Expected output:

Pipeline: clickstream_quality
Overall DQ Score: 100.0%

Common Mistakes

1. Testing Only at the End

Quality checks must run at every layer (bronze → silver → gold). A single bad record corrupts all downstream aggregations.

2. Not Testing Source Data

Assume third-party data is always unclean. Run not_null and accepted_values tests on every source immediately after ingestion.

3. No Freshness Alerts

A pipeline that loads stale data silently is worse than a broken pipeline. Always set freshness SLAs with alerts.

4. Tests That Never Fail

If all tests always pass, they're not testing enough. Review test coverage quarterly and add edge case expectations.

5. Ignoring Data Distribution Drift

Schema tests catch structural changes, but distribution drift (e.g., "status" suddenly has 50% nulls instead of 0%) often goes undetected.

Practice Questions

1. What are the six data quality dimensions? Completeness, uniqueness, timeliness, validity, accuracy, consistency.

2. How does Great Expectations differ from dbt tests? GE is declarative (assertions about data state), engine-agnostic, and runs outside the warehouse. dbt tests are SQL-based, run inside the warehouse, and integrate with dbt models.

3. What is the difference between a generic and a singular dbt test? Generic tests are defined in YAML schema files (not_null, unique, relationships). Singular tests are custom SQL queries that return failing rows.

4. What is data profiling and why is it useful? Profiling analyzes column statistics (null %, unique count, min/max, distribution) to discover data characteristics and suggest quality expectations.

5. Challenge: Design a data quality framework for a real-time pipeline processing 10M events/hour. How would sampling work? What would you alert on?

Mini Project: Data Quality Dashboard

class DQDashboard:
    def __init__(self):
        self.scores = {}
        self.alerts = []

    def record_run(self, pipeline_name, score):
        from datetime import datetime
        if pipeline_name not in self.scores:
            self.scores[pipeline_name] = []
        self.scores[pipeline_name].append((datetime.now(), score))
        if score < 80:
            self.alerts.append({"pipeline": pipeline_name, "score": score, "severity": "CRITICAL" if score < 60 else "WARNING"})

    def dashboard(self):
        print("\n=== Data Quality Dashboard ===")
        for pipeline, entries in self.scores.items():
            latest = entries[-1][1] if entries else 0
            bar = "█" * (latest // 10)
            print(f"  {pipeline}: {latest}% {bar}")
        print(f"  Active Alerts: {len(self.alerts)}")

dash = DQDashboard()
dash.record_run("clickstream", 92.1)
dash.record_run("clickstream", 73.2)
dash.record_run("orders", 95.0)
dash.dashboard()

FAQ

What is the difference between data quality testing and data monitoring?

Data quality testing validates data against expectations at a point in time (batch). Data monitoring continuously checks quality metrics over time, alerting on drifts, anomalies, and freshness violations.

How often should data quality tests run?

Run schema and completeness tests on every pipeline run. Run freshness checks continuously (every 5-15 minutes). Run distribution profiling daily or weekly to detect slow drifts.

What is data quarantine?

Data that fails quality checks is moved to a quarantine zone instead of blocking the entire pipeline. Quarantined data is logged, alerted on, and can be reprocessed after fixing the root cause.

ETL Pipelines
Data Lake vs Warehouse
dbt (Data Build Tool)

What's Next

You now understand data quality monitoring! Next, explore Data Lake vs Warehouse for architecture decisions, and learn about Apache Airflow for orchestrating quality-checked pipelines in production.

  • Practice daily — Add Great Expectations checks to your ETL pipeline
  • Build a project — Create a DQ dashboard that alerts on Slack when tests fail
  • Explore related topics — Check out Monte Carlo and Sifflet for managed data Observability

Remember: every expert was once a beginner. Keep coding!

← Previous Data Lineage — Tracking Data Flow, Impact Analysis and Governance Next → Data Lake vs Data Warehouse — Architecture Comparison

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Data Engineering