Data Quality Checks That Catch Real Bugs

Ensuring data quality is paramount in modern data platforms, especially as data pipelines become more complex and downstream systems increasingly rely on…

Ensuring data quality is paramount in modern data platforms, especially as data pipelines become more complex and downstream systems increasingly rely on accurate, timely information. This article focuses on practical, implementable data quality checks using popular tools like Great Expectations and dbt, designed to proactively identify issues before they impact business operations.

We'll cover checks for structural integrity, content validity, and freshness, emphasizing real-world scenarios that often lead to data outages and operational disruptions. The goal is to build robust data validation layers that automatically flag inconsistencies and anomalies, reducing the time to detection and resolution of data-related problems.

Establishing a Data Quality Framework

Implementing data quality checks requires a structured approach. It's not just about writing a few SQL queries; it's about integrating validation into your data lifecycle. Two prominent tools for this are Great Expectations and dbt (data build tool), both offering powerful capabilities for defining, executing, and reporting on data quality tests.

Great Expectations for Data Validation

Great Expectations provides a declarative way to define "Expectations" about your data. These expectations are assertions that data must meet. It supports various data sources (databases, data frames, files) and generates detailed data quality reports and data documentation.

Installation (Python 3.7+):

pip install great_expectations pandas sqlalchemy
great_expectations init

This initializes a Great Expectations project, creating a `great_expectations` directory with configuration files. You then connect to your data source and define Expectations.

# Example: Defining Expectations for a 'sales_data' table
# In a Python script or Jupyter notebook after connecting a datasource

import great_expectations as gx
from great_expectations.core import ExpectationConfiguration

context = gx.get_context()

# Assume 'my_postgres_datasource' is already configured
validator = context.get_validator(
    datasource_name="my_postgres_datasource",
    entity_name="default", # Or the name of your data asset
    batch_request={
        "data_asset_name": "sales_data",
        "limit": 100000, # Or a specific batch identifier
    }
)

# Add Expectations
validator.expect_table_row_count_to_be_between(min_value=10000, max_value=1000000)
validator.expect_column_to_exist(column="order_id")
validator.expect_column_values_to_be_unique(column="order_id")
validator.expect_column_values_to_not_be_null(column="customer_id")
validator.expect_column_values_to_be_between(column="order_total", min_value=0.01, max_value=100000.00)
validator.expect_column_values_to_be_in_set(column="status", value_set=["PENDING", "COMPLETED", "CANCELLED"])
validator.expect_column_max_to_be_between(column="order_date", min_value="2023-01-01", max_value=validator.get_current_datetime().strftime("%Y-%m-%d"))

# Save the Expectation Suite
validator.save_expectation_suite(discard_failed_expectations=False)

# Run a Checkpoint to validate data against the suite
checkpoint_result = context.run_checkpoint(
    checkpoint_name="my_sales_data_checkpoint",
    batch_request={
        "datasource_name": "my_postgres_datasource",
        "data_asset_name": "sales_data",
        "entity_name": "default"
    },
    expectation_suite_name="sales_data_suite"
)

# Access results: checkpoint_result.run_results

dbt Tests for Data Transformation Pipelines

dbt focuses on the "T" in ELT, primarily for transforming data in your warehouse. It has built-in testing capabilities that are excellent for ensuring the quality of your transformed models. dbt tests are SQL queries that return 0 rows for success and >0 rows for failure.

Installation (Python 3.7+):

pip install dbt-core dbt-postgres # or dbt-snowflake, dbt-bigquery, etc.
dbt init my_data_project

Defining dbt tests in schema.yml:

# models/marts/schema.yml
version: 2

models:
  - name: fact_orders
    description: Aggregated order facts
    columns:
      - name: order_id
        description: Primary key for orders
        tests:
          - unique
          - not_null
      - name: customer_id
        description: Foreign key to dim_customers
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: order_total
        description: Total amount of the order
        tests:
          - not_null
          - dbt_utils.expression_is_true: # Custom test (requires dbt-utils package)
              expression: "order_total >= 0"
          - dbt_utils.at_least_one: # Ensure there's at least one order above a threshold
              expression: "order_total > 5000"

  - name: dim_products
    description: Product dimension table
    columns:
      - name: product_id
        tests:
          - unique
          - not_null
      - name: product_name
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_match_regex: # From dbt_expectations package
              regex: "^[A-Za-z0-9\\s\\-]+$" # Alphanumeric, spaces, hyphens
              severity: warn # Failures will be warnings, not errors

Run tests with dbt test.

Essential Data Quality Checks

1. Structural Integrity Checks

These checks ensure the fundamental shape and consistency of your data.

• Row Counts: Verify that the number of rows is within an expected range. Significant deviations can indicate upstream ingestion failures or filtering issues.

  -- dbt custom test (in tests/assert_row_count_within_range.sql)
  SELECT
      count(*)
  FROM {{ model }}
  HAVING NOT (count(*) BETWEEN {{ var('min_rows', 10000) }} AND {{ var('max_rows', 1000000) }})

  Configure in schema.yml:

  models:
    - name: fact_orders
      tests:
        - assert_row_count_within_range:
            vars:
              min_rows: 50000
              max_rows: 500000

• Schema Validation: Ensure expected columns exist and data types are correct. Great Expectations excels here with expect_column_to_exist and expect_column_values_to_be_of_type. dbt's schema.yml implicitly validates column presence.
• Uniqueness Constraints: Crucial for primary keys. Use expect_column_values_to_be_unique in Great Expectations or dbt's built-in unique test.
• Not Null Constraints: For mandatory fields. Use expect_column_values_to_not_be_null or dbt's not_null test.

2. Content Validity Checks

These go deeper into the actual values within your data.

• Value Ranges/Enums: Ensure numeric values fall within a reasonable range (e.g., age > 0 and < 120), or categorical values are from a predefined set (e.g., 'status' in ['active', 'inactive']).

  -- Great Expectations example
  validator.expect_column_values_to_be_between(column="price", min_value=0.01, max_value=9999.99)
  validator.expect_column_values_to_be_in_set(column="payment_method", value_set=["CREDIT_CARD", "PAYPAL", "BANK_TRANSFER"])

• Referential Integrity: Validate foreign key relationships. dbt's relationships test is specifically designed for this. Great Expectations offers expect_column_values_to_be_in_type_list and more advanced custom expectation capabilities.

  -- dbt relationships test (from schema.yml above)
  - relationships:
      to: ref('dim_customers')
      field: customer_id

• Regex Matching: For structured strings like email addresses, product codes, or phone numbers.

  -- Using dbt_expectations package
  # models/marts/schema.yml
    - name: user_emails
      columns:
        - name: email_address
          tests:
            - dbt_expectations.expect_column_values_to_match_regex:
                regex: '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}
    
  
  

3. Freshness and Timeliness Checks

Stale data is often more detrimental than slightly incorrect data. These checks ensure data is updated as expected.

• Last Update Time: Verify that a table or partition has been updated within a specified SLA.

  -- dbt custom freshness check (in tests/assert_freshness.sql)
  SELECT 1
  FROM {{ model }}
  WHERE MAX(updated_at) < current_timestamp - interval '{{ var("freshness_hours", 2) }} hours'

  This test expects 0 rows for success. Configure as a custom test in schema.yml with `freshness_hours` variable.
• Event Time vs. Processing Time: Compare timestamps in the data (e.g., event_timestamp) with the time the data was processed or loaded (e.g., loaded_at). Large discrepancies can indicate processing delays.

4. Volume and Anomaly Detection

Significant changes in data volume often precede actual data errors. These checks don't validate correctness directly but flag suspicious patterns.

• Row Count Volatility: Monitor day-over-day or hour-over-hour row count changes. A 30% deviation (up or down) from the average of the last N periods can be a good threshold for alerting.

  -- Conceptual SQL for daily row count anomaly (requires historical data)
  WITH daily_counts AS (
      SELECT
          DATE_TRUNC('day', event_timestamp) AS day,
          COUNT(*) AS current_day_rows
      FROM fact_orders
      GROUP BY 1
  ),
  avg_counts AS (
      SELECT
          AVG(current_day_rows) AS avg_rows_last_7_days
      FROM daily_counts
      WHERE day >= current_date - interval '7 days' AND day < current_date
  )
  SELECT
      dc.day,
      dc.current_day_rows,
      ac.avg_rows_last_7_days,
      (dc.current_day_rows - ac.avg_rows_last_7_days) * 100.0 / ac.avg_rows_last_7_days AS percentage_change
  FROM daily_counts dc, avg_counts ac
  WHERE dc.day = current_date
  HAVING ABS((dc.current_day_rows - ac.avg_rows_last_7_days) * 100.0 / ac.avg_rows_last_7_days) > 30; -- Alert if change > 30%

  Many data quality platforms offer pre-built anomaly detection for metrics like row counts or column distributions.
• Distribution Skew: Monitor the distribution of key categorical columns (e.g., country_code, product_category). A sudden shift in the proportion of values could indicate upstream data corruption or changes in business logic.

Common Pitfalls and Troubleshooting

• Over-testing vs. Under-testing: Don't test every single column for every possible condition. Focus on critical columns, business-critical tables, and known historical problem areas. Start with unique, not-null, and referential integrity tests, then expand.
• Alert Fatigue: Ensure alerts are actionable and configured with appropriate severity levels. Differentiating between "warning" (e.g., product_name regex failure) and "error" (e.g., order_id not unique) is crucial. Use alerting integrations (Slack, PagerDuty) effectively.
• Data Volume Challenges: Running comprehensive checks on petabyte-scale datasets can be slow and expensive. Consider sampling strategies for non-critical checks, or run checks on recent partitions/batches only. Great Expectations and dbt both support batching and incremental testing.
• Maintenance Overhead: Data schemas and business rules evolve. Data quality tests must evolve with them. Integrate test definition and maintenance into your CI/CD pipeline for data, treating tests as code.
• False Positives/Negatives: Be prepared to tune your thresholds for volume and anomaly checks. A 30% change might be normal during a sales event or holiday period, leading to false positives. Conversely, subtle data drift might be missed by overly broad thresholds. Regularly review failed tests and adjust accordingly.
• Lack of Remediation Strategy: Detecting a data quality issue is only half the battle. Have a clear runbook for what to do when a test fails: who to notify, what dashboards to check, how to revert bad data, or how to fix upstream issues.