Great Expectations: Automated Data Quality Checks for ML Pipelines

import great_expectations as gx
import pandas as pd

context = gx.get_context()

df = pd.read_csv("orders.csv")
validator = context.sources.pandas_default.read_dataframe(df)

# Basic column expectations
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_be_unique("order_id")
validator.expect_column_values_to_be_in_set("status", ["pending", "shipped", "delivered", "cancelled"])

# Statistical expectations
validator.expect_column_mean_to_be_between("total_usd", min_value=10.0, max_value=500.0)
validator.expect_column_min_to_be_between("total_usd", min_value=0.0, max_value=1.0)
validator.expect_column_values_to_be_between("total_usd", min_value=0, max_value=10000)

# Schema expectations
validator.expect_column_to_exist("created_at")
validator.expect_table_row_count_to_be_between(min_value=1000, max_value=10_000_000)

validator.save_expectation_suite(discard_failed_expectations=False)

import great_expectations as gx

context = gx.get_context()

# PostgreSQL
pg_datasource = context.sources.add_postgres(
    name="production_db",
    connection_string="postgresql://user:pass@host:5432/dbname",
)
asset = pg_datasource.add_table_asset("orders", table_name="orders")

# BigQuery
bq_datasource = context.sources.add_bigquery(
    name="bigquery_prod",
    project="my-project",
)

checkpoint = context.add_or_update_checkpoint(
    name="daily_orders_checkpoint",
    validations=[
        {
            "batch_request": asset.build_batch_request(),
            "expectation_suite_name": "orders_suite",
        }
    ],
)

results = checkpoint.run()

if not results["success"]:
    raise ValueError("Data quality check failed  -  pipeline halted")

great_expectations docs build
great_expectations docs serve  # opens browser with validation history