Great Expectations in Production — Expectations, Checkpoints, and CI/CD Integration

Tests for Data, Not Just Code

Your application code has a test suite. Your data does not. That asymmetry is why a silent data incident — a nulled-out join key, a currency column that switched from dollars to cents, a partition that arrived half-empty — can flow through every downstream dashboard and model before anyone notices.

Great Expectations (GX) closes that gap. It lets you write assertions about data — an Expectation — in Python, group them into suites, run them as part of a pipeline, and get a human-readable report when reality drifts from what you declared. Think pytest, but the subject under test is a table or a dataframe.

This guide covers GX Core 1.x (current release 1.18 at time of writing). If you last used GX in the 0.15–0.18 era, note that the 1.0 rewrite replaced the old DataConnector / YAML-datasource machinery with a fluent, object-oriented API. Almost every snippet you find in old blog posts will not run. Everything below is written against the 1.x API.

Install and the Data Context

GX Core is a plain Python library supporting Python 3.10–3.13. Install it from PyPI. Everything starts from a Data Context — the entry point that holds your configuration, suites, and result stores.

pip install great_expectations

# in Python
python -c "import great_expectations as gx; print(gx.__version__)"

There are three flavours of context, and choosing the right one per environment matters more than it looks. A File context persists configuration as YAML on disk; an Ephemeral context lives only in memory; a Cloud context syncs to GX Cloud.

import great_expectations as gx

# Persistent: writes a gx/ folder of YAML — commit it to git.
context = gx.get_context(mode="file")

# Or point it somewhere explicit:
context = gx.get_context(mode="file", project_root_dir="./data_quality")

# In-memory: nothing survives the process. Ideal for CI and unit tests.
context = gx.get_context(mode="ephemeral")

# No mode -> GX returns a File context if one exists in the cwd
# hierarchy, otherwise an Ephemeral one.
context = gx.get_context()

The practical rule: use a file context for the durable definition of your suites (checked into version control alongside the pipeline code), and an ephemeral context in CI where you rebuild everything from that code on each run. Never let two environments quietly share one on-disk context.

Data Sources, Assets, and Batches

GX 1.x models data in three layers. A Data Source is a backend (pandas, Spark, PostgreSQL, …). A Data Asset is a table or dataframe within it. A Batch Definitiondescribes how to slice that asset into the unit you actually validate — a whole table, or one day's partition.

In-memory pandas

import great_expectations as gx
import pandas as pd

context = gx.get_context()
df = pd.read_parquet("s3://lake/events/dt=2026-07-09/part-0.parquet")

data_source = context.data_sources.add_pandas("pandas")
data_asset = data_source.add_dataframe_asset(name="events")

# Dataframes are always provided whole; the dataframe is passed
# at batch-fetch time as a batch parameter.
batch_definition = data_asset.add_batch_definition_whole_dataframe("bd")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})

A SQL table (PostgreSQL)

# Keep the credentials out of code — GX resolves ${VAR} references
# against config_variables.yml / environment variables.
data_source = context.data_sources.add_postgres(
    "warehouse",
    connection_string="postgresql+psycopg2://${DB_USER}:${DB_PASS}@${DB_HOST}/analytics",
)
data_asset = data_source.add_table_asset(name="orders", table_name="fct_orders")
batch_definition = data_asset.add_batch_definition_whole_table("bd")
batch = batch_definition.get_batch()

Spark — validate a DataFrame inside an existing job:

# Reuse the SparkSession your job already created rather than
# letting GX spin up a second context.
data_source = context.data_sources.add_spark(
    name="spark", force_reuse_spark_context=True
)
data_asset = data_source.add_dataframe_asset(name="events")
batch_definition = data_asset.add_batch_definition_whole_dataframe("bd")
batch = batch_definition.get_batch(batch_parameters={"dataframe": spark_df})

Expectations: the Actual Assertions

An Expectation is a declarative claim about data, exposed as a class under gx.expectations. The Expectation Gallery lists the full built-in catalogue — column values, aggregates, schema shape, uniqueness, set membership, and more. You can validate a single one interactively against a batch:

# Interactive check — returns a result you can inspect immediately.
result = batch.validate(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="passenger_count", min_value=1, max_value=6
    )
)
print(result.success)  # True / False

In practice you rarely validate one Expectation at a time. You collect them into an Expectation Suite — the equivalent of a test file — and register it on the context so it can be re-run, versioned, and shared.

suite = context.suites.add(
    gx.core.expectation_suite.ExpectationSuite(name="orders_contract")
)

# Structural expectations: the schema you depend on exists.
suite.add_expectation(
    gx.expectations.ExpectColumnToExist(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)

# Not-null on the columns your joins and metrics rely on.
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id")
)

# Range + set-membership: catch the "dollars became cents" class of bug.
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount_usd", min_value=0, max_value=100000
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="status", value_set=["pending", "paid", "refunded", "cancelled"]
    )
)

# Volume: an empty or truncated load is itself an incident.
suite.add_expectation(
    gx.expectations.ExpectTableRowCountToBeBetween(min_value=1000)
)

This suite is a data contract expressed as code. It says, precisely, what the orders table must look like for everything downstream to be trustworthy — and it lives in git next to the pipeline that produces the table. This pairs naturally with an explicit schema contract at the interface boundary.

Severity: Not Every Failure Should Stop the Line

A key GX 1.x addition is per-Expectation severity. A null primary key should fail the pipeline hard; a row-count that dipped 3% below yesterday probably deserves a Slack ping, not a page. Setting severity lets one suite carry both.

# A broken contract — block the pipeline.
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(
        column="order_id", severity="critical"
    )
)

# A soft anomaly — surface it, but let the data through.
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount_usd", min_value=0, max_value=100000,
        severity="warning"
    )
)

The discipline here matters more than the API. Teams that mark everything critical end up with an alert channel nobody reads; teams that mark everything a warning ship broken data. Reserve critical for the assertions whose violation genuinely means "do not publish this data" — usually keys, not-nulls, and gross volume checks — and let distributional drift be a warning your observability layer tracks over time.

Validation Definitions and Checkpoints

A suite alone is inert. To run it you bind it to data with a Validation Definition (a suite + a batch definition), then wrap one or more of those in a Checkpoint — the executable unit that validates, applies your chosen result verbosity, and fires actions.

# Bind the suite to the data it validates.
validation_definition = context.validation_definitions.add(
    gx.core.validation_definition.ValidationDefinition(
        name="orders_validation",
        data=batch_definition,
        suite=suite,
    )
)

checkpoint = context.checkpoints.add(
    gx.Checkpoint(
        name="orders_checkpoint",
        validation_definitions=[validation_definition],
        result_format={"result_format": "COMPLETE"},
    )
)

result = checkpoint.run(batch_parameters={"dataframe": df})
print(result.describe())        # readable summary
print(result.success)           # overall pass/fail

result_format controls how much detail each Validation Result carries. The default is SUMMARY; COMPLETE includes the actual unexpected values, which is what you want when a human has to debug why the check failed. Use COMPLETE in development and CI, and consider dialling it back on huge tables where the unexpected-value list itself becomes large.

Actions: Do Something When Data Breaks

The reason to use a Checkpoint over a raw batch.validate() is Actions — callbacks that run after validation. Two matter most in production: rebuilding Data Docs(GX's auto-generated HTML report of every result) and notifying a channel on failure.

from great_expectations.checkpoint import (
    SlackNotificationAction,
    UpdateDataDocsAction,
)

action_list = [
    SlackNotificationAction(
        name="notify_on_failure",
        slack_token="${validation_notification_slack_webhook}",
        slack_channel="${validation_notification_slack_channel}",
        notify_on="failure",           # all | success | failure | critical | warning | info
        show_failed_expectations=True,
    ),
    UpdateDataDocsAction(name="refresh_data_docs"),
]

checkpoint = context.checkpoints.add(
    gx.Checkpoint(
        name="orders_checkpoint",
        validation_definitions=[validation_definition],
        actions=action_list,
        result_format={"result_format": "COMPLETE"},
    )
)

notify_on="failure" keeps the channel quiet on green runs — the single most important setting for keeping alerts credible. UpdateDataDocsActionregenerates a static HTML site documenting every suite and its latest results; host that folder on S3 or an internal server and it becomes the shared source of truth for "is the data healthy right now?"

Running GX in CI/CD

The highest-leverage place to run GX is a CI pipeline that validates a sample of freshly transformed data before it is promoted — the same instinct behind data pipeline testing. Build an ephemeral context from code, run the checkpoint, and let the process exit code gate the merge.

# validate.py — invoked by CI
import sys
import great_expectations as gx
import pandas as pd

context = gx.get_context(mode="ephemeral")
# ... build data_source / asset / batch_definition / suite from code ...

checkpoint = context.checkpoints.add(
    gx.Checkpoint(
        name="ci_checkpoint",
        validation_definitions=[validation_definition],
        result_format={"result_format": "COMPLETE"},
    )
)
result = checkpoint.run(batch_parameters={"dataframe": df})

if not result.success:
    print(result.describe())
    sys.exit(1)   # fail the CI job -> block the merge

# .github/workflows/data-quality.yml
name: data-quality
on: [pull_request]
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install great_expectations pandas
      - run: python validate.py
        env:
          DB_USER: ${{ secrets.DB_USER }}
          DB_PASS: ${{ secrets.DB_PASS }}

Because the ephemeral context is rebuilt from versioned code on every run, CI can never drift from an out-of-band on-disk configuration. The suite that gates your PR is exactly the suite in the repository — no hidden state.

Orchestrating Checkpoints from Airflow

CI validates code changes; orchestration validates production data on every run. Since a Checkpoint is just a Python call, any Airflow task can run it — place a validation task between ingestion and publish so bad data is caught before it fans out.

from airflow.decorators import task

@task
def validate_orders(**context_kwargs):
    import great_expectations as gx
    ctx = gx.get_context(mode="file", project_root_dir="/opt/airflow/gx")
    checkpoint = ctx.checkpoints.get("orders_checkpoint")
    result = checkpoint.run()
    if not result.success:
        # raising fails the task -> downstream publish is skipped
        raise ValueError("Data quality gate failed: see Data Docs")
    return result.describe()

# ingest >> validate_orders() >> publish

Raising on failure turns GX into a hard gate: the publish task never runs, and your Slack action has already posted the details. GX complements rather than replaces transformation-layer tests — pair it with dbt tests (which assert inside the warehouse) by using GX at the ingestion boundary, where data first enters your control.

Production Checklist