Great Expectations with Spark and Databricks: Data Quality at Scale
Great Expectations (GX) works naturally with Pandas. The Spark integration is less documented but handles the cases that matter in production: multi-billion row Delta tables, Databricks jobs, and streaming data quality checks.
This post covers running GX against Spark DataFrames on Databricks, integrating it into your ETL pipeline, and setting up checkpoints that alert on data quality failures before corrupted data reaches downstream consumers.
Why GX on Spark
Running GX locally on a sample is fine for development. In production, you want:
- Validation against the full dataset (not a sample)
- Integration with your existing Spark jobs (no separate process)
- Checkpoints that block pipeline progress on failures
- Results stored in a data docs site your team can review
GX on Spark handles all of this.
Setting Up GX with SparkDF
pip install great_expectations pyspark# great_expectations/great_expectations.yml equivalent — programmatic setup
import great_expectations as gx
from great_expectations.datasource.fluent import SparkDatasource
context = gx.get_context() # or gx.get_context(project_root_dir="/path/to/gx")
# Register a Spark datasource
datasource = context.sources.add_spark(name="spark_source")
# Register a data asset (a specific table/path)
asset = datasource.add_dataframe_asset(name="orders_daily")Defining Expectations on a Spark DataFrame
from pyspark.sql import SparkSession
import great_expectations as gx
spark = SparkSession.builder.getOrCreate()
# Load your Spark DataFrame (could be Delta, Parquet, etc.)
df = spark.read.format("delta").load("dbfs:/mnt/data/orders/")
# Connect GX to the DataFrame
context = gx.get_context()
datasource = context.sources.add_or_update_spark("spark_datasource")
asset = datasource.add_dataframe_asset("orders")
batch_request = asset.build_batch_request(dataframe=df)
# Build expectations
suite = context.add_or_update_expectation_suite("orders_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("customer_id")
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",
["pending", "confirmed", "shipped", "cancelled", "refunded"]
)
validator.expect_table_row_count_to_be_between(min_value=1000, max_value=100_000_000)
# Save the expectation suite
validator.save_expectation_suite()Running Validation and Handling Results
def validate_orders_dataframe(df, context) -> bool:
"""Validate orders DataFrame. Returns True if all expectations pass."""
datasource = context.sources.get("spark_datasource")
asset = datasource.get_asset("orders")
batch_request = asset.build_batch_request(dataframe=df)
checkpoint = context.add_or_update_checkpoint(
name="orders_checkpoint",
validations=[{
"batch_request": batch_request,
"expectation_suite_name": "orders_suite",
}]
)
results = checkpoint.run()
# Log results
success = results.success
stats = results.run_results
for result_id, result in stats.items():
validation_result = result["validation_result"]
passed = sum(1 for r in validation_result.results if r.success)
failed = sum(1 for r in validation_result.results if not r.success)
print(f"Validation: {passed} passed, {failed} failed")
if not result["validation_result"].success:
for r in validation_result.results:
if not r.success:
print(f" FAILED: {r.expectation_config.expectation_type} "
f"on column '{r.expectation_config.kwargs.get('column', 'N/A')}'")
print(f" Details: {r.result}")
return successIntegration with Databricks ETL Jobs
Embed validation directly in your Databricks notebook or job:
# databricks_etl_job.py (runs as Databricks Job)
from pyspark.sql import SparkSession
import great_expectations as gx
from great_expectations.exceptions import ValidationError
spark = SparkSession.builder.getOrCreate()
# Step 1: Extract
raw_orders = spark.read.format("json").load("dbfs:/mnt/landing/orders/*.json")
# Step 2: Validate raw data before transformation
context = gx.get_context(project_root_dir="/dbfs/mnt/gx-project/")
if not validate_raw_orders(raw_orders, context):
# Block pipeline — bad input data
raise ValueError("Raw orders failed validation. Halting pipeline.")
# Step 3: Transform
transformed = (raw_orders
.filter("amount > 0")
.withColumn("processed_at", F.current_timestamp())
)
# Step 4: Validate transformed data
if not validate_transformed_orders(transformed, context):
raise ValueError("Transformed orders failed validation. Check transformation logic.")
# Step 5: Load
transformed.write.format("delta").mode("append").save("dbfs:/mnt/warehouse/orders/")
print("Pipeline completed successfully — all data quality checks passed.")Delta Live Tables Integration
For Delta Live Tables (DLT), GX integrates via @dlt.expect:
# DLT pipeline with built-in quality constraints
import dlt
from pyspark.sql.functions import col
@dlt.table(name="orders_validated")
@dlt.expect_or_fail("order_id is not null", "order_id IS NOT NULL")
@dlt.expect_or_fail("positive amount", "amount > 0")
@dlt.expect_or_drop("valid status",
"status IN ('pending', 'confirmed', 'shipped', 'cancelled')")
def orders_validated():
return dlt.read("orders_raw")For more complex expectations that DLT's @dlt.expect can't express, run GX as a separate validation step:
@dlt.table(name="orders_gx_validated")
def orders_gx_validated():
df = dlt.read("orders_validated")
context = gx.get_context(project_root_dir="/Volumes/catalog/schema/gx/")
if not validate_business_rules(df, context):
raise ValueError("Business rule validation failed")
return dfCustom Expectations for Business Rules
When built-in expectations don't cover your domain:
from great_expectations.expectations.expectation import ColumnPairMapExpectation
from great_expectations.execution_engine import SparkDFExecutionEngine
class ExpectColumnPairToObeyOrderConstraint(ColumnPairMapExpectation):
"""Custom expectation: end_date must be after start_date."""
map_metric = "column_pair_values.a_greater_than_b"
examples = [
{
"data": {
"start_date": ["2026-01-01", "2026-01-05"],
"end_date": ["2026-01-10", "2026-01-01"], # second row violates
},
"tests": [
{
"title": "end_date must be after start_date",
"include_in_gallery": True,
"in": {
"column_A": "end_date",
"column_B": "start_date",
},
"out": {
"success": False,
"unexpected_count": 1,
},
}
],
}
]Storing Results and Alerting
# Store results in Delta table for historical tracking
def store_validation_results(results, run_date: str):
rows = []
for result_id, result in results.run_results.items():
for expectation_result in result["validation_result"].results:
rows.append({
"run_date": run_date,
"expectation_type": expectation_result.expectation_config.expectation_type,
"column": expectation_result.expectation_config.kwargs.get("column"),
"success": expectation_result.success,
"result": str(expectation_result.result),
})
spark.createDataFrame(rows).write \
.format("delta") \
.mode("append") \
.save("dbfs:/mnt/warehouse/gx_results/")
# Alert on failures
def alert_on_failure(results, pipeline_name: str):
if not results.success:
failures = [
r for result in results.run_results.values()
for r in result["validation_result"].results
if not r.success
]
message = f"Data quality failure in {pipeline_name}:\n"
for f in failures[:10]: # first 10 failures
message += f" - {f.expectation_config.expectation_type}: {f.result}\n"
send_slack_alert(channel="#data-quality-alerts", message=message)Performance Considerations
Running GX on Spark DataFrames triggers Spark jobs. For very large tables:
# Use partial batching for expensive expectations
validator.expect_table_row_count_to_be_between(
min_value=1_000_000,
max_value=500_000_000
) # fast — COUNT(*) only
# Column-level stats (triggers full scan)
validator.expect_column_mean_to_be_between(
"amount", min_value=50, max_value=500
) # slower for 10B row tables — use sampling
# For very large tables, sample before validation
sample_df = full_df.sample(fraction=0.01, seed=42)
asset.build_batch_request(dataframe=sample_df)GX on Spark gives you the same data quality assertions as the Pandas version, but running against the actual production data at scale. The checkpoint pattern ensures your pipeline is blocked before bad data propagates downstream.
