for-quality

---
name: for-quality
description: Generates a data quality test notebook for any Fabric Lakehouse table — row counts, null ratios, value ranges, uniqueness, freshness, schema drift, simple statistical checks. Use this skill whenever the user wants to add tests, validations, quality gates, freshness checks, or anomaly thresholds anywhere in a Fabric pipeline. Trigger on phrases like "test this", "quality check", "validate", "freshness", "row count check", "schema test", "uniqueness", "data drift", "DQ", or any question about verifying data integrity between layers. Should be invoked after any bronze / silver / gold table is built or modified.
license: MIT
---

# Fabric Data Quality

Generates a PySpark notebook that runs a battery of data quality checks against any Lakehouse table. Quality checks are first-class citizens: every table in the data product has a paired DQ notebook.

## Preconditions

Before generating the notebook, confirm with the user:

1. **Target table** — the table being validated
2. **Layer** — bronze / silver / gold (determines default check set)
3. **Business key** — for uniqueness checks (silver / gold only)
4. **SLA** — max acceptable lag in hours for freshness check
5. **Critical columns** — columns where nulls are forbidden
6. **Numeric ranges** — `{column: (min, max)}` for value range checks
7. **Failure mode** — `warn` (log only) or `fail` (raise + block downstream)

If any of these is missing, ask before generating.

## Default check sets per layer

**Bronze checks** (minimum)
- Row count > 0 over the last ingestion window
- `_ingested_at` freshness within SLA
- No fully-null columns

**Silver checks** (standard)
- All bronze checks +
- Strict schema match against declared types
- Business key uniqueness (zero duplicates)
- Critical columns non-null
- Numeric columns within declared ranges
- Reject ratio under threshold (e.g., < 1%)

**Gold checks** (consumption-ready)
- All silver checks +
- Row count vs expected grain (e.g., one row per cell per hour)
- No gaps in time partitions over the SLA window
- Aggregate sanity: known totals match expectations (where applicable)

## Rules — what DQ notebooks MUST do

1. Run each check independently. One check failing does not skip the others.
2. Collect all results into a single DataFrame, write to `LH_${env}.ops.dq_results` with run metadata.
3. Print a clear human-readable summary at the end.
4. Exit with a non-zero status (raise `AssertionError`) if any `fail`-mode check failed.
5. Be idempotent and read-only — never mutate the target table.

## Rules — what DQ notebooks MUST NOT do

- Don't write to the target table.
- Don't mark a check as passed if it could not be evaluated (e.g., source missing). Use `unknown` instead.
- Don't hide check definitions in helper functions without a docstring.

## Notebook template

```python
# ---
# table: <layer>.<table_name>__dq
# layer: ops
# owner: <team>
# sources: [<target_table>]
# target: LH_${env}.ops.dq_results
# refresh: <after every load>
# ---

# %% [Imports]
from pyspark.sql import functions as F
from datetime import datetime, timezone
import time

# %% [Parameters]
env = "dev"
target_table   = f"LH_{env}.<layer>.<table_name>"
results_table  = f"LH_{env}.ops.dq_results"
business_key   = ["<key_col_1>", "<key_col_2>"]
freshness_col  = "<event_at | _ingested_at>"
freshness_sla_hours = 2
critical_cols  = ["<col_1>", "<col_2>"]
ranges         = {"prb_util": (0.0, 100.0), "rsrp": (-140.0, -40.0)}
failure_mode   = "fail"   # "fail" or "warn"

run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
df = spark.table(target_table)
results = []

def record(name: str, status: str, value, threshold=None, detail=""):
    results.append({
        "run_id": run_id,
        "table": target_table,
        "check": name,
        "status": status,                    # pass | fail | warn | unknown
        "value": float(value) if value is not None else None,
        "threshold": float(threshold) if threshold is not None else None,
        "detail": detail,
        "checked_at": datetime.now(timezone.utc),
    })

# %% [Check 1 — row count > 0]
n = df.count()
record("row_count_positive", "pass" if n > 0 else "fail", n, 0)

# %% [Check 2 — freshness within SLA]
max_ts = df.agg(F.max(freshness_col)).collect()[0][0]
if max_ts is None:
    record("freshness", "unknown", None, freshness_sla_hours, "no timestamp found")
else:
    lag_hours = (datetime.now(timezone.utc) - max_ts.replace(tzinfo=timezone.utc)).total_seconds() / 3600
    record("freshness", "pass" if lag_hours <= freshness_sla_hours else "fail",
           lag_hours, freshness_sla_hours, f"lag={lag_hours:.2f}h")

# %% [Check 3 — business key uniqueness]
if business_key:
    dup_count = (
        df.groupBy(*business_key).count().filter("count > 1").count()
    )
    record("uniqueness", "pass" if dup_count == 0 else "fail", dup_count, 0,
           f"duplicate keys={dup_count}")

# %% [Check 4 — critical columns non-null]
for c in critical_cols:
    null_ratio = df.filter(F.col(c).isNull()).count() / max(n, 1)
    record(f"non_null:{c}", "pass" if null_ratio == 0 else "fail",
           null_ratio, 0.0, f"null_ratio={null_ratio:.4f}")

# %% [Check 5 — numeric ranges]
for col, (lo, hi) in ranges.items():
    bad = df.filter((F.col(col) < lo) | (F.col(col) > hi)).count()
    record(f"range:{col}", "pass" if bad == 0 else "fail",
           bad, 0, f"out_of_range={bad} (allowed=[{lo},{hi}])")

# %% [Persist results]
df_results = spark.createDataFrame(results)
df_results.write.format("delta").mode("append").saveAsTable(results_table)

# %% [Summary]
failed = [r for r in results if r["status"] == "fail"]
warned = [r for r in results if r["status"] == "warn"]
print(f"[dq] {target_table} | total={len(results)} | failed={len(failed)} | warned={len(warned)}")
for r in failed:
    print(f"  FAIL: {r['check']} → value={r['value']} threshold={r['threshold']} {r['detail']}")

# %% [Fail loudly if configured]
if failed and failure_mode == "fail":
    raise AssertionError(f"{len(failed)} DQ check(s) failed on {target_table}")
```

## Output format

When invoked, produce:
1. The notebook at `notebooks/dq/<table_name>__dq.py`
2. A summary listing every check generated, its threshold, and its failure mode
3. A recommendation on which checks to mark `fail` vs `warn`

## Handoff

After DQ is in place, suggest:
> "Quality gates are in place. Want me to chain this notebook into the orchestration pipeline so silver / gold only refresh if upstream DQ passes?"