mo
f36c8906bc
Adds native Command Center features (no new containers) integrated as sub-tabs in the existing Data Explorer and Data Quality views: - Continuous Data Quality (dq_monitor.py): live completeness/uniqueness/validity/ freshness scorecards via Trino with rolling trends → DataQuality "Live Monitoring". - Ownership & stewardship (catalog_governance.py): owner/steward/tier matrix, orphan detection, business glossary; local store best-effort synced to OpenMetadata (owner PATCH) → Data Explorer "Ownership". - Access & policy posture: per-dataset compliance combining PII masking, ownership, live DQ and observability alerts vs data contracts → Data Explorer "Access & Policies". - Lineage (lineage.py): staged source→CDC→Spark→S3→Iceberg→Trino→serving graph with live row counts and column-level PII/masking tracing → Data Explorer "Lineage". - Observability (observability.py): volume/freshness/schema-drift monitoring with alerts → Data Explorer "Observability". - Shared lake_meta.py dataset registry + bounded Trino client; fast native row-count and PK-indexed freshness so monitors stay cheap on 25-54M-row tables. - LLM context (lab_context.py) enriched with DQ scores, ownership and active alerts.
346 lines
13 KiB
Python
346 lines
13 KiB
Python
"""Continuous data-quality monitoring (Disease #2 — Dirty Data).
|
||
|
||
Where dq-api runs Great-Expectations/Soda on *uploaded* files, this agent runs
|
||
quality checks continuously against the *live* business tables through Trino and
|
||
produces a per-dataset scorecard (0-100) across five dimensions:
|
||
|
||
completeness – non-null ratio across columns
|
||
uniqueness – distinct/key ratio (duplicate detection)
|
||
validity – domain rules (non-negative amounts, present timestamps)
|
||
freshness – age of the newest record vs a threshold
|
||
volume – row count + delta vs the previous cycle
|
||
|
||
A rolling score history powers trend sparklines. The loop streams the exact
|
||
SQL it runs into the Data Custodian terminal so operators can see the checks.
|
||
|
||
Endpoints:
|
||
GET /api/dq/scorecards -> all datasets, dimensions, score, trend
|
||
GET /api/dq/scorecard/{key} -> one dataset detail
|
||
POST /api/dq/run -> run one cycle now
|
||
POST /api/dq/config -> {enabled, interval_s, sample}
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import threading
|
||
import time
|
||
from collections import deque
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Body
|
||
from fastapi.responses import JSONResponse
|
||
|
||
from lake_meta import DATASETS, DATASET_BY_KEY, trino, trino_scalar, discover_columns
|
||
|
||
router = APIRouter(prefix="/api/dq", tags=["data-quality"])
|
||
|
||
FRESHNESS_THRESHOLD_MIN = 60.0 # newest row older than this => freshness degraded
|
||
MAX_COLS = 24
|
||
|
||
_lock = threading.Lock()
|
||
_state: dict[str, Any] = {
|
||
"enabled": True,
|
||
"interval_s": 180.0,
|
||
"sample": 20000,
|
||
"running": False,
|
||
"cycles": 0,
|
||
"last_cycle_ts": 0.0,
|
||
"next_run_ts": 0.0,
|
||
"started_at": None,
|
||
"cards": {}, # key -> latest scorecard
|
||
"history": {d["key"]: deque(maxlen=60) for d in DATASETS},
|
||
"cols_cache": {}, # key -> [{name,type}]
|
||
"feed": deque(maxlen=40),
|
||
}
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def _term(text: str, level: str = "info", phase: str = "dq") -> None:
|
||
try:
|
||
from agent_terminal import emit_threadsafe
|
||
emit_threadsafe("data-custodian", text, level=level, phase=phase)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _feed(text: str, level: str = "info") -> None:
|
||
_state["feed"].appendleft({"ts": _now().isoformat(), "text": text, "level": level})
|
||
|
||
|
||
def _score_color(score: float) -> str:
|
||
if score >= 90:
|
||
return "#34d399"
|
||
if score >= 75:
|
||
return "#fbbf24"
|
||
if score >= 50:
|
||
return "#fb923c"
|
||
return "#f87171"
|
||
|
||
|
||
def _obs_volume_freshness(key: str) -> tuple[int | None, float | None]:
|
||
"""Read row count + freshness from the observability monitor (in-memory)."""
|
||
try:
|
||
from observability import _state, _lock
|
||
with _lock:
|
||
st = _state["datasets"].get(key) or {}
|
||
return st.get("rows"), st.get("freshness_age_min")
|
||
except Exception:
|
||
return None, None
|
||
|
||
|
||
def _columns(ds: dict[str, Any]) -> list[dict[str, str]]:
|
||
key = ds["key"]
|
||
cols = _state["cols_cache"].get(key)
|
||
if not cols:
|
||
cols = discover_columns(ds)
|
||
if cols:
|
||
_state["cols_cache"][key] = cols
|
||
return cols or []
|
||
|
||
|
||
def _assess(ds: dict[str, Any], sample: int) -> dict[str, Any]:
|
||
key = ds["key"]
|
||
fq = ds["fqtn"]
|
||
cols = _columns(ds)
|
||
col_names = [c["name"] for c in cols][:MAX_COLS]
|
||
dims: dict[str, Any] = {}
|
||
issues: list[str] = []
|
||
col_profiles: list[dict[str, Any]] = []
|
||
|
||
# ── one bounded query: completeness + uniqueness + validity over a sample ──
|
||
has_unique_key = ds.get("unique_key", True)
|
||
key_col = ds.get("key_col") if ds.get("key_col") in col_names else (col_names[0] if col_names else None)
|
||
if not has_unique_key:
|
||
key_col = None # no single-column unique key (e.g. composite PK) → skip dedup
|
||
ts_col = ds.get("ts_col") if ds.get("ts_col") in col_names else None
|
||
amt_col = ds.get("amount_col") if ds.get("amount_col") in col_names else None
|
||
|
||
selects = ["count(*) AS n"]
|
||
for i, c in enumerate(col_names):
|
||
selects.append(f'count("{c}") AS c{i}')
|
||
if key_col:
|
||
selects.append(f'count(DISTINCT "{key_col}") AS dk')
|
||
if amt_col:
|
||
selects.append(f'count_if("{amt_col}" >= 0) AS amt_ok')
|
||
if ts_col:
|
||
selects.append(f'count_if("{ts_col}" IS NOT NULL) AS ts_ok')
|
||
|
||
sql = f"SELECT {', '.join(selects)} FROM (SELECT * FROM {fq} LIMIT {sample}) s"
|
||
_term(f"$ trino: profile {key} ({len(col_names)} cols · sample {sample:,})", level="cmd", phase="dq")
|
||
cols_out, rows = trino(sql, timeout=40.0)
|
||
rec = dict(zip(cols_out, rows[0])) if rows else {}
|
||
n = int(rec.get("n") or 0)
|
||
|
||
if n > 0:
|
||
# completeness
|
||
non_null_ratios = []
|
||
for i, c in enumerate(col_names):
|
||
cnt = int(rec.get(f"c{i}") or 0)
|
||
ratio = cnt / n
|
||
non_null_ratios.append(ratio)
|
||
col_profiles.append({"name": c, "completeness": round(ratio * 100, 1),
|
||
"nulls": n - cnt})
|
||
completeness = round(100.0 * sum(non_null_ratios) / max(1, len(non_null_ratios)), 1)
|
||
dims["completeness"] = completeness
|
||
worst = sorted(col_profiles, key=lambda x: x["completeness"])[:3]
|
||
for w in worst:
|
||
if w["completeness"] < 95:
|
||
issues.append(f'{w["name"]} {100 - w["completeness"]:.0f}% null')
|
||
|
||
# uniqueness / dedup
|
||
if key_col and rec.get("dk") is not None:
|
||
dk = int(rec["dk"])
|
||
uniq = round(100.0 * dk / n, 1)
|
||
dims["uniqueness"] = uniq
|
||
dups = n - dk
|
||
if dups > 0:
|
||
issues.append(f"{dups:,} duplicate {key_col} in sample")
|
||
|
||
# validity
|
||
valid_parts = []
|
||
if amt_col and rec.get("amt_ok") is not None:
|
||
valid_parts.append(int(rec["amt_ok"]) / n)
|
||
if ts_col and rec.get("ts_ok") is not None:
|
||
valid_parts.append(int(rec["ts_ok"]) / n)
|
||
if valid_parts:
|
||
validity = round(100.0 * sum(valid_parts) / len(valid_parts), 1)
|
||
dims["validity"] = validity
|
||
if validity < 99 and amt_col:
|
||
issues.append(f"negative/invalid {amt_col}")
|
||
|
||
# ── volume + freshness: reuse the observability monitor's in-memory probes
|
||
# (it already polls count(*) and max(ts)). We deliberately do NOT issue our
|
||
# own count/max here — those are expensive full scans on 25-54M-row tables —
|
||
# so a DQ cycle stays fast and the score is driven by the cheap sample. ──
|
||
volume, fresh_age_min = _obs_volume_freshness(key)
|
||
if volume is None:
|
||
volume = n # sample size until observability reports the real count
|
||
if fresh_age_min is not None:
|
||
fresh_score = 100.0 if fresh_age_min <= FRESHNESS_THRESHOLD_MIN else max(
|
||
0.0, 100.0 - (fresh_age_min - FRESHNESS_THRESHOLD_MIN) / 5.0)
|
||
dims["freshness"] = round(fresh_score, 1)
|
||
if fresh_age_min > FRESHNESS_THRESHOLD_MIN and not ds.get("curated"):
|
||
issues.append(f"stale {fresh_age_min:.0f}m")
|
||
|
||
score = round(sum(dims.values()) / len(dims), 1) if dims else 0.0
|
||
prev = _state["cards"].get(key, {})
|
||
prev_vol = prev.get("volume")
|
||
delta = (volume - prev_vol) if (volume is not None and prev_vol is not None) else None
|
||
|
||
card = {
|
||
"key": key, "label": ds["label"], "engine": ds["engine"], "color": ds["color"],
|
||
"table": fq, "domain": ds.get("domain"),
|
||
"score": score, "score_color": _score_color(score),
|
||
"dimensions": dims, "volume": volume, "volume_delta": delta,
|
||
"freshness_age_min": round(fresh_age_min, 1) if fresh_age_min is not None else None,
|
||
"issues": issues[:5], "columns": len(col_names),
|
||
"worst_columns": sorted(col_profiles, key=lambda x: x["completeness"])[:5],
|
||
"ts": _now().isoformat(),
|
||
}
|
||
lvl = "ok" if score >= 90 else ("warn" if score >= 60 else "err")
|
||
_term(f" ← {key}: score {score} · " + " · ".join(f"{k} {v}" for k, v in dims.items())
|
||
+ (f" · {len(issues)} issue(s)" if issues else ""), level=lvl, phase="dq")
|
||
return card
|
||
|
||
|
||
def run_cycle() -> dict[str, Any]:
|
||
if _state["running"]:
|
||
return {"ok": True, "skipped": "already running"}
|
||
_state["running"] = True
|
||
sample = int(_state["sample"])
|
||
n_ok = 0
|
||
try:
|
||
_term(f"═══ DQ monitor cycle {_state['cycles'] + 1} — live tables via Trino ═══",
|
||
level="info", phase="cycle")
|
||
for ds in DATASETS:
|
||
if not _state["enabled"]:
|
||
break
|
||
try:
|
||
card = _assess(ds, sample)
|
||
except Exception as exc:
|
||
card = {"key": ds["key"], "label": ds["label"], "engine": ds["engine"],
|
||
"color": ds["color"], "table": ds["fqtn"], "score": None,
|
||
"score_color": "#64748b", "dimensions": {}, "error": str(exc)[:160],
|
||
"issues": [f"check failed: {str(exc)[:80]}"], "ts": _now().isoformat()}
|
||
_term(f" ✗ {ds['key']}: {str(exc)[:120]}", level="err", phase="dq")
|
||
with _lock:
|
||
_state["cards"][ds["key"]] = card
|
||
if card.get("score") is not None:
|
||
_state["history"][ds["key"]].append({"t": _now().strftime("%H:%M"), "score": card["score"]})
|
||
n_ok += 1
|
||
_state["cycles"] += 1
|
||
_state["last_cycle_ts"] = time.time()
|
||
_state["next_run_ts"] = time.time() + float(_state["interval_s"])
|
||
avg = _overall_score()
|
||
_feed(f"cycle {_state['cycles']} — {n_ok}/{len(DATASETS)} datasets scored · platform DQ {avg}")
|
||
_term(f"═══ cycle {_state['cycles']} done — platform DQ score {avg} ═══", level="ok", phase="cycle")
|
||
finally:
|
||
_state["running"] = False
|
||
return {"ok": True, "scored": n_ok, "platform_score": _overall_score()}
|
||
|
||
|
||
def _overall_score() -> float | None:
|
||
vals = [c["score"] for c in _state["cards"].values() if c.get("score") is not None]
|
||
return round(sum(vals) / len(vals), 1) if vals else None
|
||
|
||
|
||
def _loop() -> None:
|
||
_state["started_at"] = _now().isoformat()
|
||
time.sleep(20)
|
||
while True:
|
||
try:
|
||
if _state["enabled"]:
|
||
run_cycle()
|
||
except Exception:
|
||
pass
|
||
time.sleep(max(30.0, float(_state["interval_s"])))
|
||
|
||
|
||
threading.Thread(target=_loop, daemon=True, name="dq-monitor").start()
|
||
|
||
|
||
def scorecards_view() -> dict[str, Any]:
|
||
with _lock:
|
||
cards = []
|
||
for ds in DATASETS:
|
||
c = _state["cards"].get(ds["key"])
|
||
if c:
|
||
cards.append({**c, "trend": list(_state["history"][ds["key"]])})
|
||
else:
|
||
cards.append({"key": ds["key"], "label": ds["label"], "engine": ds["engine"],
|
||
"color": ds["color"], "table": ds["fqtn"], "score": None,
|
||
"score_color": "#64748b", "dimensions": {}, "issues": [],
|
||
"trend": [], "pending": True})
|
||
# platform dimension averages
|
||
dim_avg: dict[str, list[float]] = {}
|
||
for c in cards:
|
||
for k, v in (c.get("dimensions") or {}).items():
|
||
dim_avg.setdefault(k, []).append(v)
|
||
return {
|
||
"ok": True,
|
||
"enabled": _state["enabled"],
|
||
"interval_s": _state["interval_s"],
|
||
"sample": _state["sample"],
|
||
"running": _state["running"],
|
||
"cycles": _state["cycles"],
|
||
"last_cycle_ts": _state["last_cycle_ts"],
|
||
"next_run_ts": _state["next_run_ts"],
|
||
"platform_score": _overall_score(),
|
||
"dimension_averages": {k: round(sum(v) / len(v), 1) for k, v in dim_avg.items()},
|
||
"cards": cards,
|
||
"feed": list(_state["feed"])[:20],
|
||
}
|
||
|
||
|
||
def summary_for_llm() -> dict[str, Any]:
|
||
with _lock:
|
||
return {
|
||
"platform_dq_score": _overall_score(),
|
||
"datasets": [{"key": c["key"], "score": c.get("score"),
|
||
"issues": c.get("issues", []), "freshness_age_min": c.get("freshness_age_min")}
|
||
for c in _state["cards"].values()],
|
||
}
|
||
|
||
|
||
@router.get("/scorecards")
|
||
async def get_scorecards() -> JSONResponse:
|
||
return JSONResponse(scorecards_view())
|
||
|
||
|
||
@router.get("/scorecard/{key}")
|
||
async def get_scorecard(key: str) -> JSONResponse:
|
||
with _lock:
|
||
c = _state["cards"].get(key)
|
||
if not c:
|
||
return JSONResponse({"ok": False, "error": "unknown or not yet scored"}, status_code=404)
|
||
return JSONResponse({"ok": True, "card": {**c, "trend": list(_state["history"].get(key, []))}})
|
||
|
||
|
||
@router.post("/run")
|
||
async def post_run() -> JSONResponse:
|
||
from starlette.concurrency import run_in_threadpool
|
||
res = await run_in_threadpool(run_cycle)
|
||
return JSONResponse({**res, "scorecards": scorecards_view()})
|
||
|
||
|
||
@router.post("/config")
|
||
async def post_config(body: dict = Body(default={})) -> JSONResponse:
|
||
if "enabled" in body:
|
||
_state["enabled"] = bool(body["enabled"])
|
||
if "interval_s" in body:
|
||
try:
|
||
_state["interval_s"] = max(30.0, min(1800.0, float(body["interval_s"])))
|
||
except Exception:
|
||
pass
|
||
if "sample" in body:
|
||
try:
|
||
_state["sample"] = max(1000, min(200000, int(body["sample"])))
|
||
except Exception:
|
||
pass
|
||
return JSONResponse({"ok": True, "enabled": _state["enabled"],
|
||
"interval_s": _state["interval_s"], "sample": _state["sample"]})
|