Great Expectations: Data Quality Validation & Testing Guide

Great Expectations is the leading open-source Python library for data quality validation. Define a contract for your data, validate every pipeline run against it, generate human-readable documentation, and catch bad data before it reaches production systems.

What is Great Expectations?

Great Expectations (GE) is an open-source Python library for defining, documenting, and enforcing data quality expectations in data pipelines. Instead of writing ad-hoc assertions scattered across pipeline code, GE provides a structured framework: define Expectation Suites (quality contracts), connect them to data via Validators, and run them at checkpoints in your pipeline. Failed validations can block pipeline execution or trigger alerts before bad data propagates downstream.

GE works with Pandas DataFrames, Spark DataFrames, SQL databases (via SQLAlchemy), and files (CSV, Parquet, JSON). The Data Docs feature auto-generates an HTML documentation site showing all expectations, their validation history, and observed values — making data quality transparent to all stakeholders without requiring access to the pipeline code.

Creating an Expectation Suite

# pip install great-expectations
import great_expectations as gx

# Initialize a Data Context (creates gx/ config directory)
context = gx.get_context()

# Connect to a Pandas datasource
datasource = context.sources.add_pandas("my_datasource")
asset = datasource.add_csv_asset("orders", filepath_or_buffer="orders.csv")
batch_request = asset.build_batch_request()

# Create an Expectation Suite
suite = context.add_expectation_suite("orders_suite")

# Get a Validator (loads a batch + attaches the suite)
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="orders_suite",
)

# Define expectations
validator.expect_column_to_exist("order_id")
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_be_unique("order_id")
validator.expect_column_values_to_not_be_null("amount")
validator.expect_column_values_to_be_between("amount", min_value=0, max_value=1_000_000)
validator.expect_column_values_to_be_in_set(
    "status", ["placed", "shipped", "delivered", "returned"]
)
validator.expect_column_values_to_match_regex(
    "email", r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+.[a-zA-Z0-9-.]+$"
)
validator.expect_table_row_count_to_be_between(min_value=1000, max_value=10_000_000)

# Save expectations to the suite
validator.save_expectation_suite(discard_failed_expectations=False)
print("Suite saved with", len(suite.expectations), "expectations")

Checkpoints, Data Docs & Alerts

# Checkpoint — run validations and build Data Docs
import great_expectations as gx

context = gx.get_context()

# Add a checkpoint that links a suite to a datasource
checkpoint = context.add_checkpoint(
    name="orders_checkpoint",
    validations=[
        {
            "batch_request": {
                "datasource_name": "my_datasource",
                "data_asset_name": "orders",
            },
            "expectation_suite_name": "orders_suite",
        }
    ],
    action_list=[
        {
            "name": "store_validation_result",
            "action": {"class_name": "StoreValidationResultAction"},
        },
        {
            "name": "update_data_docs",
            "action": {"class_name": "UpdateDataDocsAction"},
        },
        {
            "name": "send_slack_notification",
            "action": {
                "class_name": "SlackNotificationAction",
                "slack_webhook": "https://hooks.slack.com/...",
                "notify_on": "failure",
            },
        },
    ],
)

# Run the checkpoint
result = context.run_checkpoint(checkpoint_name="orders_checkpoint")

if not result["success"]:
    raise ValueError("Data validation FAILED — blocking pipeline execution!")

# Open Data Docs in browser
context.open_data_docs()

Integrating GE into an Airflow Pipeline

# Integrate Great Expectations into an Airflow DAG
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import great_expectations as gx

def validate_data(**context):
    gx_context = gx.get_context(
        context_root_dir="/opt/airflow/gx"
    )
    result = gx_context.run_checkpoint(
        checkpoint_name="orders_checkpoint",
        batch_parameters={
            "partition_date": context["ds"]  # pass Airflow execution date
        },
    )
    if not result["success"]:
        failed = [
            r["expectation_config"]["expectation_type"]
            for r in result["run_results"].values()
            for r in r["validation_result"]["results"]
            if not r["success"]
        ]
        raise ValueError(f"Validation failed for: {failed}")

with DAG("etl_with_validation", schedule_interval="@daily",
         start_date=datetime(2024, 1, 1), catchup=False) as dag:
    validate = PythonOperator(
        task_id="validate_orders",
        python_callable=validate_data,
    )
    transform = PythonOperator(task_id="transform", python_callable=run_transform)
    validate >> transform  # block transform if validation fails

Great Expectations vs dbt Tests vs Soda