airflow-dq by example — a 15-minute tour¶

Note

This page mirrors examples/README.md in the repository — if the two ever disagree, the repo copy next to the code wins. There is also a standalone, PyPI-based deployment example under examples/pypi-deployment/ — see Deployment.

Everything below runs against the bundled demo stack. The five example DAGs under dags/examples/ auto-load into it (paused, tagged examples), and each one's docstring says what it teaches, how to run it, and what to look at afterwards.

cp .env.example .env          # defaults are fine for the demo
docker compose up -d          # Airflow 3.1 standalone + seeded Postgres
make ui-build                 # once, if ui/dist doesn't exist yet
docker compose logs airflow | grep "Password for user 'admin'"   # login credentials

Airflow UI: http://localhost:8080 → log in → Data Quality in the nav. For curl, the demo's legacy API token works as a Bearer token (change-me-in-prod unless you set DQ_API_TOKEN); in production you'd use your Airflow JWT instead.

TOKEN="change-me-in-prod"
DQ="http://localhost:8080/dq/api"

1. The UI in two minutes¶

Overview — one card per contract: latest status counts, pass rate, SLA badges, anomaly count. Click a card to drill in.
Quality Dashboard — KPIs, quality-over-time (red points = server-flagged anomalous runs), checks by dimension, and the results table; failing checks expand to the actual offending rows. Filter by dag to separate gates from monitors.
Contract Editor — author no-code (introspect columns, pick rules, live-Test against the real table before saving) or import from dbt; saves are versioned and breaking changes are classified.
Quarantine — every quarantined row with filters + bulk resolve.

2. Run the examples (in order)¶

DAG	Teaches
`example_01_quickstart_gate`	The 5-line integration: `bronze >> build_contract_checks("orders@v3", on_fail="fail") >> publish`. Version pinning, gate semantics (only `blocking` rules can fail a gate).
`example_02_quarantine_triage`	`on_fail="quarantine"`: bad rows are copied aside, the pipeline continues. Graded tolerances (warn/fail thresholds) instead of binary red/green. Then: Quarantine tab → select → Mark resolved.
`example_03_incremental_and_freshness`	Partition-scoped checks via `where: "created_at >= DATE '{{ data_interval_start }}'"` (backfill-safe) and the two freshness modes (`max_age` vs `row_age`) side by side.
`example_04_sla_and_anomaly`	SLAs declared in the contract (`slaProperties` → compiled `sla.*` checks) + z-score anomaly detection with a bot-excluding baseline. Run it ~5× and the trend chart starts flagging metric-level anomalies.
`example_05_payments_producer` / `_monitor`	The gate/monitor duality: an Asset-triggered, record-only monitor that audits every table update without touching producer code.

Unpause a DAG in the Airflow UI (or docker exec <airflow> airflow dags unpause <dag_id>), let it run, then open the Data Quality views it points you at.

Example 03 wants a scheduled run

Do not trigger example_03_incremental_and_freshness manually: manual runs have no data interval, so {{ data_interval_start }} cannot be rendered. airflow-dq fails closed on that — the scoped check reports status error instead of silently scanning the whole table. Unpausing the @daily DAG auto-creates the latest scheduled run.

3. Author a contract in the editor¶

Contract Editor → type a table name (try payments) → Load columns → ✨ Profile & suggest drafts checks from the live data → adjust → Test any block to see pass/fail + offending rows before saving → set id/version → Save contract. Saving diffs against the previous version and classifies the change (red = breaking); versions are immutable — the editor auto-bumps to the next free number.

4. Import your dbt tests¶

examples/dbt_schema.yml is a typical dbt file. Either paste it into the editor's Import from dbt panel, or:

python -c 'import json,sys; print(json.dumps({"schema_yml": open("examples/dbt_schema.yml").read()}))' \
  | curl -s -X POST "$DQ/contracts/import/dbt" \
      -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" -d @- | python -m json.tool

You get a contract draft (not_null→required+notNull, unique, accepted_values, relationships→referentialIntegrity); anything unmappable is listed in skipped_tests, not silently dropped.

5. Alerting that doesn't page you 20 times¶

One grouped alert per (run, contract), routed by contract owner and severity — set in .env and restart:

# every failure to one channel:
DQ_ALERT_WEBHOOK=https://hooks.slack.com/services/T000/B000/XXXX
# or per-owner, severity-aware routing (fail vs infra-error), any mix of channels:
DQ_ALERT_ROUTES='{"growth@example.com": {"fail": "https://hooks.slack.com/services/...",
                                          "error": "pagerduty://YOUR_ROUTING_KEY"}}'
DQ_BASE_URL=http://localhost:8080     # makes alert links clickable

Channels by URL scheme: https://… webhook (Slack-compatible), mailto:… (SMTP via DQ_SMTP_*), pagerduty://<routing_key>, opsgenie://<api_key>.

6. Contracts as code: propose instead of save¶

With the contracts dir in git and DQ_CONTRACTS_GIT=true, ticking Propose as PR in the editor commits the new version to a dq/contract-<id>-v<N> branch instead of activating it — review the classified diff in your normal PR flow, merge to ship.

7. API quick reference¶

curl -s -H "Authorization: Bearer $TOKEN" "$DQ/overview"                          # per-contract rollup
curl -s -H "Authorization: Bearer $TOKEN" "$DQ/results?contract=payments@v1&status=fail&limit=20"
curl -s -H "Authorization: Bearer $TOKEN" "$DQ/trends?contract=web_events@v1"     # incl. anomalous flag
curl -s -H "Authorization: Bearer $TOKEN" "$DQ/quarantine?resolved=false"
curl -s -X POST -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
     -d '{"ids": [1, 2], "resolved_by": "me"}' "$DQ/quarantine/resolve"
curl -s -H "Authorization: Bearer $TOKEN" "$DQ/contracts/payments/versions"       # version timeline

The full route list lives in the REST API reference.

Where to go next¶

Wire your own table: copy example_01, point it at your contract ref and an Airflow Connection id as backend (falls back to DQ_DWH_DSN).
The configuration reference has every knob; the rules reference every rule's exact semantics; architecture the design decisions.