Connecting GX to DuckDB

43.1 A SQL Data Source for DuckDB (23 min)

GX connects to SQL databases through SQLAlchemy connection strings. For DuckDB, the format is:

duckdb:////absolute/path/to/macro.duckdb      (four slashes = absolute path)
duckdb:///relative/path/macro.duckdb          (three slashes = relative path)

Keep the path out of the project file

Hard-coding a path in great_expectations.yml would break on another machine. GX substitutes ${VARIABLE} placeholders from environment variables, or from gx/uncommitted/config_variables.yml (never committed). It's the same idea as dlt's secrets on Day 23.

export MACRO_DUCKDB_URL="duckdb:////workspaces/macro-data-course/macro.duckdb"

To make it permanent, add that line to ~/.bashrc, or create a Codespaces secret named MACRO_DUCKDB_URL. For databases with passwords (Postgres, Snowflake, …), this mechanism is how credentials stay out of Git.

Register the data source

Create quality/day43_connect.py:

from gx_helpers import get_context

context = get_context()

try:
    ds = context.data_sources.get("macro_duckdb")
except Exception:
    ds = context.data_sources.add_sql(
        name="macro_duckdb",
        connection_string="${MACRO_DUCKDB_URL}",
    )

print(ds)

source .venv-gx/bin/activate
python quality/day43_connect.py
grep -A3 macro_duckdb quality/gx/great_expectations.yml

The YAML stores the placeholder, not the path. Commit it safely.

One writer at a time, again. GX only reads, but DuckDB still needs other processes not to hold a write lock. Run GX after ingestion and dbt have finished, never at the same time. On Days 48–50 the pipeline enforces this order. If you see a lock error, close any open DuckDB CLI session or running dbt command.

43.2 Table Assets at Every Layer (25 min)

A table asset points GX at one table or view. Add assets for each layer we want to protect:

ASSETS = [
    # name in GX                 schema          table
    ("raw_statcan_observations", "raw_dlt",      "statcan_observations"),
    ("raw_statcan_cpi_full",     "raw_dlt",      "statcan_cpi_full"),
    ("stg_statcan__cpi",         "staging",      "stg_statcan__cpi"),
    ("int_macro__observations",  "intermediate", "int_macro__observations"),
    ("macro_monthly",            "marts",        "macro_monthly"),
    ("fct_open_economy_daily",   "marts",        "fct_open_economy_daily"),
]

for asset_name, schema, table in ASSETS:
    try:
        ds.get_asset(asset_name)
    except Exception:
        ds.add_table_asset(name=asset_name, schema_name=schema, table_name=table)
    print("✓", asset_name)

Why validate at several layers?

Validating raw data is the heart of Days 48–50: problems are cheapest to catch before they flow through dozens of models.

Your first SQL-backed validation

import great_expectations as gx

asset = ds.get_asset("macro_monthly")
try:
    bd = asset.get_batch_definition("full")
except Exception:
    bd = asset.add_batch_definition_whole_table(name="full")

batch = bd.get_batch()
print(batch.head())

result = batch.validate(gx.expectations.ExpectColumnValuesToBeBetween(
    column="unemployment_rate", min_value=0, max_value=100))
print("Unemployment 0–100%:", result.success, result.result.get("unexpected_count"))

GX translates the expectation into SQL and DuckDB does the work: the table never gets loaded into Python. That's why SQL data sources scale to large tables.

Validating raw text data

raw_dlt.statcan_cpi_full stores every column as text (Day 18: raw layers never guess types). Expectations still work, and a raw asset is the right place to check formats:

raw = ds.get_asset("raw_statcan_cpi_full")
try:
    raw_bd = raw.get_batch_definition("full")
except Exception:
    raw_bd = raw.add_batch_definition_whole_table(name="full")
raw_batch = raw_bd.get_batch()

for exp in [
    gx.expectations.ExpectColumnValuesToNotBeNull(column="ref_date"),
    gx.expectations.ExpectColumnValueLengthsToEqual(column="ref_date", value=7),   # 'YYYY-MM'
    gx.expectations.ExpectTableColumnsToMatchSet(
        column_set=["ref_date", "geo", "products_and_product_groups", "uom",
                    "vector", "value", "status"],
        exact_match=False),                                   # at least these columns exist
]:
    r = raw_batch.validate(exp)
    print(f"{'✓' if r.success else '✗'} {exp.type}")

The column-set expectation is a contract on the provider's schema: if StatCan renames a column, GX reports it at the boundary, before dbt fails in a confusing way three models later.

43.3 Query Assets: Validating Derived Views (23 min)

Sometimes the thing you want to validate isn't a table but a question: "the unemployment series only", or "the months where the two CPI paths disagree". A query asset wraps a SQL query.

One series from the long table

UNEMP_SQL = """
    select obs_date, value
    from intermediate.int_macro__observations
    where series_name = 'unemployment_rate'
"""
try:
    unemp = ds.get_asset("unemployment_rate")
except Exception:
    unemp = ds.add_query_asset(name="unemployment_rate", query=UNEMP_SQL)

The consistency trick: validate a query of violations

Recall the "test = query returning violations" pattern (Days 18 and 36). In GX, wrap the violations query as an asset and expect zero rows:

CPI_MISMATCH_SQL = """
    select v.obs_date, v.value as cpi_vector, t.cpi as cpi_table
    from intermediate.int_macro__observations as v
    join staging.stg_statcan__cpi as t
      on t.month = v.obs_date and t.geo_code = 'CA' and t.component = 'All-items'
    where v.series_name = 'cpi_all_items'
      and abs(v.value - t.cpi) > 1e-9
"""
try:
    mismatch = ds.get_asset("cpi_vector_vs_table_mismatches")
except Exception:
    mismatch = ds.add_query_asset(name="cpi_vector_vs_table_mismatches", query=CPI_MISMATCH_SQL)

bd = mismatch.add_batch_definition_whole_table(name="full")
result = bd.get_batch().validate(gx.expectations.ExpectTableRowCountToEqual(
    value=0, meta={"plan_id": "Q04", "dimension": "consistency", "severity": "critical"}))
print("CPI paths consistent:", result.success)

This implements plan check Q04 from Day 41, a check that no dbt test covered.

Query assets vs dbt models. If a derived view is useful beyond quality checks, it belongs in dbt (tested, documented, on the lineage graph). Query assets are best for quality-specific questions like "rows that violate X", which nobody else consumes.

43.4 Batch Definitions: Validating One Month at a Time (23 min)

A whole-table batch validates all history on every run. That's fine for small tables, but often you want to validate only the newest data, the month that just arrived, for three reasons:

1. Speed: checking one month is cheaper than 50 years.
2. Signal: a failure in the latest batch is news; a failure in 1987 is old news you've already accepted (with mostly).
3. Accountability: results per batch build a validation history (Data Docs, Days 44–47).

GX can partition a table asset by a date column:

asset = ds.get_asset("stg_statcan__cpi")
try:
    monthly_bd = asset.get_batch_definition("by_month")
except Exception:
    monthly_bd = asset.add_batch_definition_monthly(name="by_month", column="month")

# Validate one specific month
batch = monthly_bd.get_batch(batch_parameters={"year": 2026, "month": 7})
print(batch.head())
print(batch.validate(gx.expectations.ExpectColumnValuesToBeBetween(
    column="cpi", min_value=0, max_value=1000)).success)

Without batch_parameters, get_batch() returns the most recent partition, which is exactly what a daily pipeline wants.

Try it 43.4

Add a monthly batch definition to the unemployment_rate query asset (column obs_date) and validate the latest month: value between 0 and 100 and not null. Then validate April 2020 explicitly. What's special about it?

unemp = ds.get_asset("unemployment_rate")
try:
    ubd = unemp.get_batch_definition("by_month")
except Exception:
    ubd = unemp.add_batch_definition_monthly(name="by_month", column="obs_date")

for params in [None, {"year": 2020, "month": 4}]:
    b = ubd.get_batch(batch_parameters=params) if params else ubd.get_batch()
    r = b.validate(gx.expectations.ExpectColumnValuesToBeBetween(
        column="value", min_value=0, max_value=100))
    print(params or "latest", r.success, b.head()["value"].tolist())

43.5 SQL vs pandas: Where the Work Happens (15 min)

You now know two ways to point GX at DuckDB data:

The fallback pattern

If a particular expectation isn't supported for DuckDB in your GX version, or you hit a connection issue, validate a DataFrame instead:

import duckdb
import great_expectations as gx
from gx_helpers import get_context, PROJECT_ROOT

context = get_context()
con = duckdb.connect(str(PROJECT_ROOT.parent / "macro.duckdb"), read_only=True)
df = con.sql("select month, headline_yoy, unemployment_rate from marts.macro_monthly").df()
con.close()

try:
    src = context.data_sources.get("pandas_fallback")
except Exception:
    src = context.data_sources.add_pandas(name="pandas_fallback")
try:
    asset = src.get_asset("macro_monthly_df")
except Exception:
    asset = src.add_dataframe_asset(name="macro_monthly_df")
try:
    bd = asset.get_batch_definition("whole")
except Exception:
    bd = asset.add_batch_definition_whole_dataframe(name="whole")

batch = bd.get_batch(batch_parameters={"dataframe": df})
print(batch.validate(gx.expectations.ExpectColumnValuesToMatchRegex(
    column="month", regex=r"^\d{4}-\d{2}-01")).success)   # dates as text would need casting first

read_only=True means this script can't accidentally modify the warehouse. Use it for every quality script that opens DuckDB directly.

(The regex example is illustrative. month is a date, so for a format check you'd first convert it to a string, e.g. df["month"].astype(str). Try it.)

43.6 Capstone: setup_gx.py + smoke_check.py (28 min)

Build two scripts that turn today's experiments into a reproducible foundation:

• quality/setup_gx.py: idempotent (safe to run many times). It registers the data source, all assets and batch definitions. Run it whenever the setup changes.
• quality/smoke_check.py: runs a handful of critical expectations from the quality plan against production data, prints a report, and exits non-zero on any failure.

This mirrors dbt's split between project definition and running (dbt build), and Day 23's split between configuration and execution.

Requirements for smoke_check.py (plan IDs from Day 41):

"""Idempotently register the DuckDB data source, assets and batch definitions."""
from gx_helpers import get_context

TABLE_ASSETS = [
    ("raw_statcan_observations", "raw_dlt",      "statcan_observations"),
    ("raw_statcan_cpi_full",     "raw_dlt",      "statcan_cpi_full"),
    ("stg_statcan__cpi",         "staging",      "stg_statcan__cpi"),
    ("int_macro__observations",  "intermediate", "int_macro__observations"),
    ("macro_monthly",            "marts",        "macro_monthly"),
    ("fct_open_economy_daily",   "marts",        "fct_open_economy_daily"),
]

QUERY_ASSETS = {
    "unemployment_rate": """
        select obs_date, value
        from intermediate.int_macro__observations
        where series_name = 'unemployment_rate'
    """,
    "cpi_vector_vs_table_mismatches": """
        select v.obs_date, v.value as cpi_vector, t.cpi as cpi_table
        from intermediate.int_macro__observations as v
        join staging.stg_statcan__cpi as t
          on t.month = v.obs_date and t.geo_code = 'CA' and t.component = 'All-items'
        where v.series_name = 'cpi_all_items' and abs(v.value - t.cpi) > 1e-9
    """,
}

MONTHLY_PARTITIONS = {"stg_statcan__cpi": "month", "unemployment_rate": "obs_date"}


def main():
    context = get_context()

    try:
        ds = context.data_sources.get("macro_duckdb")
    except Exception:
        ds = context.data_sources.add_sql(name="macro_duckdb",
                                          connection_string="${MACRO_DUCKDB_URL}")

    for name, schema, table in TABLE_ASSETS:
        try:
            asset = ds.get_asset(name)
        except Exception:
            asset = ds.add_table_asset(name=name, schema_name=schema, table_name=table)
        ensure_batch_definitions(asset, MONTHLY_PARTITIONS.get(name))

    for name, sql in QUERY_ASSETS.items():
        try:
            asset = ds.get_asset(name)
        except Exception:
            asset = ds.add_query_asset(name=name, query=sql)
        ensure_batch_definitions(asset, MONTHLY_PARTITIONS.get(name))

    print("✓ GX setup complete:",
          sorted(a.name for a in ds.assets))


def ensure_batch_definitions(asset, month_column=None):
    try:
        asset.get_batch_definition("full")
    except Exception:
        asset.add_batch_definition_whole_table(name="full")
    if month_column:
        try:
            asset.get_batch_definition("by_month")
        except Exception:
            asset.add_batch_definition_monthly(name="by_month", column=month_column)


if __name__ == "__main__":
    main()

"""Run critical smoke checks from the data quality plan against production data."""
import sys

import great_expectations as gx

from gx_helpers import get_context

E = gx.expectations

CHECKS = [
    # (plan_id, asset, batch_definition, expectation)
    ("Q01", "unemployment_rate", "full",
     E.ExpectColumnValuesToBeBetween(column="value", min_value=0, max_value=100)),
    ("Q04", "cpi_vector_vs_table_mismatches", "full",
     E.ExpectTableRowCountToEqual(value=0)),
    ("Q07", "macro_monthly", "full",
     E.ExpectColumnValuesToBeUnique(column="month")),
    ("Q07", "macro_monthly", "full",
     E.ExpectColumnValuesToNotBeNull(column="month")),
    ("Q08", "fct_open_economy_daily", "full",
     E.ExpectColumnValuesToBeBetween(column="usd_cad", min_value=0.8, max_value=2.0)),
    ("Q09", "stg_statcan__cpi", "by_month",            # latest month only
     E.ExpectColumnValuesToBeBetween(column="cpi", min_value=0, strict_min=True)),
    ("Q10", "raw_statcan_cpi_full", "full",
     E.ExpectTableColumnsToMatchSet(
         column_set=["ref_date", "geo", "products_and_product_groups", "uom",
                     "vector", "value", "status"], exact_match=False)),
]


def main():
    context = get_context()
    ds = context.data_sources.get("macro_duckdb")
    failures = 0

    print(f"{'ID':<5} {'asset':<34} {'expectation':<42} result")
    for plan_id, asset_name, bd_name, exp in CHECKS:
        batch = ds.get_asset(asset_name).get_batch_definition(bd_name).get_batch()
        r = batch.validate(exp, result_format="SUMMARY")
        detail = r.result.get("unexpected_count", r.result.get("observed_value", ""))
        print(f"{plan_id:<5} {asset_name:<34} {exp.type:<42} "
              f"{'✓' if r.success else '✗'} {detail}")
        failures += 0 if r.success else 1

    print(f"\n{failures} failing check(s)")
    sys.exit(1 if failures else 0)


if __name__ == "__main__":
    main()

Run them:

source .venv-gx/bin/activate
python quality/setup_gx.py
python quality/setup_gx.py          # second run: nothing new created (idempotent)
python quality/smoke_check.py; echo "exit code: $?"

Break it on purpose

Prove the checks work before trusting them:

1. In smoke_check.py, temporarily set Q08's max_value to 1.2, run, observe the failure and the count of offending days, then restore it.
2. Consistency failure: in a DuckDB session (with nothing else running), create a temporary copy that disagrees, point a test query asset at it, and confirm ExpectTableRowCountToEqual(0) fails. Then drop the copy. Never edit production tables to test.

Recap: Days 41–43

Looking ahead: Days 44–47

• Expectation suites per asset, organised by the quality plan, with meta linking every expectation to its plan ID, dimension and severity.
• Profiling datasets to derive sensible expectations (ranges, distributions, allowed values) from history.
• Validation definitions and checkpoints with actions.
• Automated Data Docs: a browsable website of expectations and validation results for economists and managers, not just engineers.

Day 43 checklist

• GX connects to DuckDB through a ${MACRO_DUCKDB_URL} placeholder; no paths or secrets are committed
• I registered table assets in raw_dlt, staging, intermediate and marts, and know why each layer matters
• I use query assets for quality-specific questions, including a zero-rows consistency check
• I can validate the latest month, or any chosen month, with monthly batch definitions
• I know when to use SQL vs pandas data sources, and have a read-only fallback
• setup_gx.py is idempotent; smoke_check.py enforces critical plan checks with an exit code

git add quality/ .gitignore
git status                          # quality/gx/uncommitted must NOT be listed
git commit -m "Day 43: GX connected to DuckDB, assets, batch definitions, smoke checks"
git push