Testing

Great Expectations in CI/CD: Automated Data Quality Gates for Every Pipeline Run

HelpMeTest

HelpMeTest

— 5 min read
Great Expectations in CI/CD: Automated Data Quality Gates for Every Pipeline Run

Data quality issues in production are expensive. A corrupt batch silently flows into your model, your dashboard, or your customer reports — and you find out hours or days later. The fix: automated data quality gates in your CI/CD pipeline that fail the job before bad data goes anywhere.

Great Expectations (GX) is the standard Python library for this. This guide covers how to build GX suites, checkpoints, and integrate them into GitHub Actions so your pipeline won't deploy unless the data passes.

Why CI/CD Needs Data Quality Gates

Code CI already catches logic bugs. But data pipelines have a second class of failures that code tests can't catch:

  • Schema drift: upstream system adds or renames a column
  • Distribution shift: a vendor sends you values in a different range or unit
  • Referential integrity: foreign keys that no longer exist in the dimension table
  • Completeness: a batch that's missing 30% of expected rows

These bugs don't raise exceptions. They flow silently downstream. GX gives you a structured way to express and enforce your data contracts.

Installing Great Expectations

pip install great-expectations

Initialize a GX project (creates a great_expectations/ directory):

great_expectations init

This creates:

great_expectations/
  great_expectations.yml        # project config
  expectations/                 # expectation suite files
  checkpoints/                  # checkpoint configs
  uncommitted/
    config_variables.yml        # secrets (not committed)
    validations/                # validation results

Creating Expectation Suites

An Expectation Suite is a named collection of expectations about a dataset. Create one programmatically:

# scripts/create_suite.py
import great_expectations as gx
import pandas as pd

context = gx.get_context()

# Load a sample of your data
df = pd.read_parquet("data/events_sample.parquet")

# Create a batch from the DataFrame
datasource = context.sources.add_pandas("my_datasource")
data_asset = datasource.add_dataframe_asset("events")
batch_request = data_asset.build_batch_request(dataframe=df)

# Create the suite
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="events.basic_quality",
)

# Add expectations
validator.expect_table_row_count_to_be_between(min_value=1000, max_value=10_000_000)
validator.expect_column_to_exist("event_id")
validator.expect_column_to_exist("user_id")
validator.expect_column_to_exist("event_type")
validator.expect_column_to_exist("timestamp")
validator.expect_column_values_to_not_be_null("event_id")
validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_be_in_set(
    "event_type",
    ["click", "view", "purchase", "add_to_cart", "checkout"],
)
validator.expect_column_values_to_be_between(
    "amount",
    min_value=0.01,
    max_value=10_000.0,
    mostly=0.95,  # Allow 5% outside range (nulls from non-purchase events)
)
validator.expect_column_values_to_match_strftime_format(
    "timestamp",
    strftime_format="%Y-%m-%d %H:%M:%S",
)

# Save the suite
validator.save_expectation_suite(discard_failed_expectations=False)
print("Suite saved.")

Run it once to create the great_expectations/expectations/events.basic_quality.json file.

Writing Suites Without a Sample Dataset

If you don't have sample data in CI, define expectations directly in JSON:

// great_expectations/expectations/events.basic_quality.json
{
  "expectation_suite_name": "events.basic_quality",
  "expectations": [
    {
      "expectation_type": "expect_column_to_exist",
      "kwargs": { "column": "event_id" }
    },
    {
      "expectation_type": "expect_column_values_to_not_be_null",
      "kwargs": { "column": "event_id" }
    },
    {
      "expectation_type": "expect_column_values_to_be_in_set",
      "kwargs": {
        "column": "event_type",
        "value_set": ["click", "view", "purchase", "add_to_cart"]
      }
    },
    {
      "expectation_type": "expect_table_row_count_to_be_between",
      "kwargs": { "min_value": 1, "max_value": 10000000 }
    }
  ],
  "meta": {}
}

Creating a Checkpoint

A Checkpoint wires a batch request (data source + asset) to an expectation suite and produces a validation result. Create one:

# scripts/create_checkpoint.py
import great_expectations as gx

context = gx.get_context()

checkpoint = context.add_or_update_checkpoint(
    name="events_quality_gate",
    validations=[
        {
            "batch_request": {
                "datasource_name": "my_datasource",
                "data_asset_name": "events",
            },
            "expectation_suite_name": "events.basic_quality",
        }
    ],
)
print("Checkpoint created.")

Run the checkpoint:

result = context.run_checkpoint(checkpoint_name="events_quality_gate")

if not result.success:
    raise ValueError(f"Data quality check failed: {result}")

Integrating Into Your Pipeline Code

The cleanest approach: add a quality gate function your pipeline calls before proceeding:

# my_pipeline/quality.py
import great_expectations as gx
import pandas as pd
from typing import Optional

def assert_data_quality(df: pd.DataFrame, suite_name: str, datasource_name: str = "pipeline_ds") -> None:
    """
    Run GX validation on a DataFrame. Raises ValueError if validation fails.
    
    Args:
        df: The DataFrame to validate
        suite_name: Name of the expectation suite to use
        datasource_name: GX datasource name
    
    Raises:
        ValueError: If any expectations fail
    """
    context = gx.get_context()
    
    datasource = context.sources.add_or_update_pandas(datasource_name)
    data_asset = datasource.add_dataframe_asset("batch")
    batch_request = data_asset.build_batch_request(dataframe=df)
    
    validator = context.get_validator(
        batch_request=batch_request,
        expectation_suite_name=suite_name,
    )
    
    results = validator.validate()
    
    if not results.success:
        failed = [
            r for r in results.results if not r.success
        ]
        messages = [f"  - {r.expectation_config.expectation_type}: {r.result}" for r in failed]
        raise ValueError(f"Data quality failed ({len(failed)} expectations):\n" + "\n".join(messages))

Use it in your pipeline:

# my_pipeline/pipeline.py
from my_pipeline.quality import assert_data_quality
from my_pipeline.extract import read_events_from_s3
from my_pipeline.transform import process_events

def run_pipeline(s3_path: str) -> None:
    # Extract
    raw = read_events_from_s3(s3_path)
    
    # Quality gate on raw data
    assert_data_quality(raw, suite_name="events.basic_quality")
    
    # Transform
    processed = process_events(raw)
    
    # Quality gate on transformed data
    assert_data_quality(processed, suite_name="events.processed_quality")
    
    # Load
    load_to_warehouse(processed)

Testing the Quality Gate Itself

Write pytest tests that verify your quality gate catches bad data:

# tests/test_quality_gate.py
import pytest
import pandas as pd
from my_pipeline.quality import assert_data_quality

@pytest.fixture
def valid_events():
    return pd.DataFrame({
        "event_id": [1, 2, 3],
        "user_id": [101, 102, 103],
        "event_type": ["click", "view", "purchase"],
        "timestamp": ["2024-01-01 10:00:00", "2024-01-01 10:05:00", "2024-01-01 10:10:00"],
        "amount": [None, None, 49.99],
    })

def test_valid_events_pass_quality_gate(valid_events):
    # Should not raise
    assert_data_quality(valid_events, suite_name="events.basic_quality")

def test_missing_column_fails_quality_gate(valid_events):
    bad_df = valid_events.drop(columns=["event_id"])
    
    with pytest.raises(ValueError, match="Data quality failed"):
        assert_data_quality(bad_df, suite_name="events.basic_quality")

def test_invalid_event_type_fails_quality_gate(valid_events):
    bad_df = valid_events.copy()
    bad_df.loc[0, "event_type"] = "unknown_event"
    
    with pytest.raises(ValueError, match="Data quality failed"):
        assert_data_quality(bad_df, suite_name="events.basic_quality")

def test_null_primary_key_fails_quality_gate(valid_events):
    bad_df = valid_events.copy()
    bad_df.loc[0, "event_id"] = None
    
    with pytest.raises(ValueError, match="Data quality failed"):
        assert_data_quality(bad_df, suite_name="events.basic_quality")

GitHub Actions Integration

Add a data quality step to your CI workflow:

# .github/workflows/pipeline-test.yml
name: Pipeline CI

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v4
      
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      
      - name: Install dependencies
        run: pip install -r requirements.txt great-expectations pytest
      
      - name: Run unit tests
        run: pytest tests/unit/ -v
      
      - name: Validate expectation suites are present
        run: |
          python -c "
          import great_expectations as gx
          context = gx.get_context()
          suites = context.list_expectation_suite_names()
          print('Suites found:', suites)
          required = ['events.basic_quality', 'events.processed_quality']
          for suite in required:
              assert suite in suites, f'Missing suite: {suite}'
          print('All required suites present.')
          "
      
      - name: Run data quality integration tests
        run: pytest tests/test_quality_gate.py -v

Handling GX in a Monorepo or Lambda

In serverless or containerized environments, you often can't run the full GX context. A lightweight alternative: write thin validation functions using pandas directly, tested with pytest:

def validate_events_schema(df: pd.DataFrame) -> list[str]:
    """Returns list of validation errors. Empty = pass."""
    errors = []
    required_cols = ["event_id", "user_id", "event_type", "timestamp"]
    
    for col in required_cols:
        if col not in df.columns:
            errors.append(f"Missing column: {col}")
    
    if "event_id" in df.columns and df["event_id"].isnull().any():
        errors.append("event_id has null values")
    
    if "event_type" in df.columns:
        valid_types = {"click", "view", "purchase", "add_to_cart"}
        invalid = set(df["event_type"].unique()) - valid_types
        if invalid:
            errors.append(f"Invalid event_types: {invalid}")
    
    if len(df) == 0:
        errors.append("DataFrame is empty")
    
    return errors

This gives you testable data quality logic with zero GX overhead. Use GX when you need the full suite management, documentation, and observability features. Use plain Python validation functions when you need lightweight checks that run in tight environments.

Expectation Best Practices

Start with schema expectations. Column existence and type checks catch 80% of schema drift.

Use mostly for probabilistic expectations. mostly=0.99 means "99% of values must pass." Useful for columns that legitimately have some nulls or outliers.

Profile first, set expectations second. Run validator.expect_column_values_to_be_between using your actual data's P1/P99 range, not a guess.

Version your suites in Git. The great_expectations/expectations/ directory should be committed. Treat expectation changes like code changes.

Don't block on warnings. Separate critical expectations (block on failure) from informational ones (log but don't fail the job).

Conclusion

Great Expectations in CI/CD is a data quality gate that runs on every pipeline execution. It catches schema drift, null explosions, and out-of-range values before they corrupt downstream consumers. Wire it into your GitHub Actions workflow, write pytest tests for the gate itself, and commit your expectation suites to version control — and you'll have a feedback loop that catches data bugs at the source.

Read more

Start now free