Great Expectations Data Quality Validation¶

import great_expectations as gx
from great_expectations.core.batch import RuntimeBatchRequest
import pandas as pd

# Get GX context
context = gx.get_context(context_root_dir="validation/great_expectations")

# Load data
df = pd.read_parquet("sample-data/bronze/slot_telemetry.parquet")

# Create batch request
batch_request = RuntimeBatchRequest(
    datasource_name="casino_pandas",
    data_connector_name="runtime_data_connector",
    data_asset_name="slot_data",
    runtime_parameters={"batch_data": df},
    batch_identifiers={"default_identifier_name": "validation_run"}
)

# Run checkpoint with runtime batch
results = context.run_checkpoint(
    checkpoint_name="slot_machine_checkpoint",
    validations=[{
        "batch_request": batch_request,
        "expectation_suite_name": "slot_machine_suite"
    }]
)

# Check results
if not results.success:
    print("Validation failed!")
    for result in results.run_results.values():
        print(result)

# In Fabric Notebook
import great_expectations as gx
from pyspark.sql import SparkSession

# Get Spark session (already available in Fabric)
spark = SparkSession.builder.getOrCreate()

# Read from Lakehouse
df = spark.read.format("delta").table("bronze.slot_telemetry")

# Get GX context
context = gx.get_context()

# Create validator
validator = context.get_validator(
    datasource_name="casino_delta",
    data_connector_name="runtime_data_connector",
    data_asset_name="slot_telemetry",
    expectation_suite_name="slot_machine_suite",
    batch_identifiers={"table_name": "slot_telemetry", "layer": "bronze"},
    batch_data=df
)

# Run validation
results = validator.validate()
print(f"Validation passed: {results.success}")

# plugins/custom_expectations.py
from great_expectations.expectations.expectation import ColumnMapExpectation

class ExpectColumnValuesToBeValidMachineId(ColumnMapExpectation):
    """Expect column values to be valid casino machine IDs."""

    map_metric = "column_values.match_regex"
    success_keys = ("regex",)
    default_kwarg_values = {
        "regex": r"^(SLOT|TBL)-[A-Z0-9]{2,4}-[0-9]{3,4}$"
    }

    library_metadata = {
        "maturity": "experimental",
        "tags": ["casino", "machine_id"],
    }