Validating with GX Core

Validating BookNest's Iceberg Tables with GX Core

GX has no Iceberg 129 connector, so the script reads the table through DuckDB 61,228 into pandas 16,086 (at scale, pass a Spark 129 DataFrame from lake.py). The suite includes a freshness rule: the latest order must be from this week.

gx_orders.py: an expectation suite and checkpoint for the orders tablePython

"""GX Core 1.23: an expectation suite for the Iceberg orders table, run by a checkpoint."""
import great_expectations as gx
import great_expectations.expectations as gxe
from lakeduck import connect
orders = connect().sql("SELECT * FROM orders").df()                # 100,000 rows, via DuckDB
context = gx.get_context(mode="file", project_root_dir="gx")       # suites, results, Data Docs
batch = (context.data_sources.add_or_update_pandas("lakehouse")
         .add_dataframe_asset("orders").add_batch_definition_whole_dataframe("all"))
suite = context.suites.add_or_update(gx.ExpectationSuite(name="orders_suite", expectations=[
    gxe.ExpectTableRowCountToBeBetween(min_value=90_000, max_value=200_000),
    gxe.ExpectColumnValuesToBeUnique(column="order_id"),
    gxe.ExpectColumnValuesToBeInSet(column="status", value_set=[
        "pending", "paid", "shipped", "delivered", "returned", "cancelled"]),
    gxe.ExpectColumnValuesToBeBetween(column="discount", min_value=0, max_value=20,
                                      mostly=0.999),          # 99.9% of rows is enough
    gxe.ExpectColumnMaxToBeBetween(column="order_ts", min_value="2026-10-01T00:00:00+00:00"),
]))
checkpoint = context.checkpoints.add_or_update(gx.Checkpoint(
    name="orders_checkpoint", actions=[gx.checkpoint.UpdateDataDocsAction(name="docs")],
    validation_definitions=[context.validation_definitions.add_or_update(
        gx.ValidationDefinition(name="orders_daily", data=batch, suite=suite))]))
result = list(checkpoint.run(batch_parameters={"dataframe": orders}).run_results.values())[0]
for r in result.results:
    cfg, res = r.expectation_config, r.result
    seen = (f"observed {res['observed_value']}" if "observed_value" in res
            else f"unexpected {res['unexpected_count']}")
    column = cfg.kwargs.get("column", "")
    print(f"{'PASS' if r.success else 'FAIL'} {cfg.type:<36} {column:<11} {seen}")
print("suite success:", result.success)

Output

PASS expect_table_row_count_to_be_between             observed 100000
PASS expect_column_values_to_be_unique    order_id    unexpected 0
PASS expect_column_values_to_be_in_set    status      unexpected 0
PASS expect_column_values_to_be_between   discount    unexpected 13
FAIL expect_column_max_to_be_between      order_ts    observed 2026-06-30 23:55:02+00:00
suite success: False

Thirteen discounts above 20 stay within the 0.1% tolerance. The freshness check fails: the history ends on 30 June, the silent staleness a dashboard never shows. The checkpoint rebuilt Data Docs, here filtered to failures:

GX Core Data Docs: the orders validation, filtered to its failed freshness expectation

Keep the suites (JSON under gx/) in Git 1,932 , and let the orchestrator fail the run when result.success is false.

MENU