Self-learning DQX pipeline on Databricks

Contents

00What you build 01Workspace setup 02The config file 03Profile the table 04Load your context 05Build the RAG index 06Retrieve per table 07Generate the ruleset 08Validate and repair 09Execute with DQX 10Score 11The report 12Persist and trend 13Wire the notebook 14Failure modes

00What you are building

You add a table to a YAML file. You run a notebook. You get a hundred data quality rules that understand your domain, executed against the full table, with a score and a report.

The interesting part is not that a model can write SQL. It is that generated SQL is untrusted code about to run against production data, so most of the engineering is in the stages around generation: giving the model real statistics instead of just a schema, retrieving your own documentation so the rules encode business logic rather than generic null checks, and validating every expression four different ways before it touches a full table.

Stages 1, 4 and 6 are where the quality comes from. Stage 5, the part that feels like the product, is the least interesting one in the whole pipeline.

01Workspace setup

Compute

Serverless notebooks or a cluster on a recent DBR with Unity Catalog enabled. Serverless is convenient because the notebook’s Databricks token is available to the process automatically, which is how the embedding calls authenticate without any extra secret.

A Volume to hold the index

The retrieval index is a single JSONL file. It does not need a vector database. A Unity Catalog Volume is enough, and it gives you governance and a path that is stable across cluster restarts.

CREATE CATALOG IF NOT EXISTS main;
CREATE SCHEMA  IF NOT EXISTS main.dq;
CREATE VOLUME  IF NOT EXISTS main.dq.dq_assets;

Libraries

Pin DQX, because the check metadata schema is what your generated contracts have to match.

%pip install databricks-labs-dqx==0.15.0 langchain-openai pyyaml
dbutils.library.restartPython()

LLM endpoint credentials

Any OpenAI-compatible chat endpoint works: OpenAI, Azure OpenAI, a self-hosted gateway, or a Databricks serving endpoint. Put the three values in a secret scope rather than the notebook.

import os

os.environ["LLM_BASE_URL"] = dbutils.secrets.get("dq", "llm_base_url")
os.environ["LLM_API_KEY"]  = dbutils.secrets.get("dq", "llm_api_key")
os.environ["LLM_MODEL"]    = "gpt-4o"        # or your gateway's model name

Embeddings are separate and stay inside Databricks. databricks-gte-large-en is a pay-per-token Foundation Model endpoint that every workspace has, it returns 1024-dimensional vectors, and on serverless it authenticates with the notebook’s own token.

Code layout

Keep the pipeline in a repo synced as a Git folder and import it from the notebook. Notebook cells are a bad place for four hundred lines of repair logic.

dq_pipeline/
├── configs/dq_config.yaml
├── data/context/
│   ├── default/                 # loaded for every table
│   └── <app>/<entity>/           # loaded for one table
├── src/dq/
│   ├── profile.py               # stage 1
│   ├── context.py               # stage 2
│   ├── rag.py                   # stages 3 and 4
│   ├── generate.py              # stage 5
│   ├── repair.py                # stage 6
│   ├── execute.py               # stage 7
│   ├── score.py                 # stage 8
│   └── report.py                # stage 9
└── notebooks/run_dq.py

02The config file drives everything

Onboarding a new table should be one YAML block and nothing else. If it needs a code change, the design is wrong.

configs/dq_config.yaml

rag:
  enabled: true
  top_k: 12
  chunk_words: 250
  chunk_overlap: 50
  volume_path: /Volumes/main/dq/dq_assets
  index_file: rag_index.jsonl

job:
  output_catalog: main
  output_schema: dq

sources:
  - id: sales_customers                     # unique run identifier
    application: sales                      # used in the retrieval query
    entity: customer
    context_directory: data/context/sales/customer   # or null
    pk_column: ["customer_id"]              # list: composite keys supported
    table:
      type: table
      location: main.sales.customers        # Unity Catalog FQN
    llm:
      max_rules: 50

Three things here matter more than they look.

• application and entity are not decoration. They go into the retrieval query, so they steer which documentation chunks come back.
• pk_column as a list means composite keys work without a special case, and the duplicate-rate calculation reads it directly.
• max_rules is a hard cap. Without one you get either four rules or four hundred, depending on the model’s mood.

03Profile the table in one pass

A model given only a schema writes generic rules. A model given the real shape of the data writes specific ones.

The difference is one aggregate pass. Build every column’s statistics in a single Spark action rather than looping, and use approximate distinct counts, because an exact COUNT(DISTINCT) across two hundred columns is a shuffle you do not need.

src/dq/profile.py

from pyspark.sql import DataFrame
from pyspark.sql import functions as F
from pyspark.sql.types import StringType

def build_table_pack(df: DataFrame, table_fqn: str, row_count: int,
                     max_columns: int = 200) -> dict:
    """Bounded column stats, safe to put in a prompt. No raw row values."""
    fields = df.schema.fields[:max_columns]
    agg_exprs = []

    for i, field in enumerate(fields):
        col = F.col(f"`{field.name}`")
        # treat blank strings as missing, not just NULL
        if isinstance(field.dataType, StringType):
            is_bad = col.isNull() | (F.length(F.trim(col.cast("string"))) == 0)
            agg_exprs.append(F.approx_count_distinct(col).alias(f"_dist_{i}"))
        else:
            is_bad = col.isNull()
        agg_exprs.append(F.sum(F.when(is_bad, 1).otherwise(0)).alias(f"_null_{i}"))

    row = df.agg(*agg_exprs).collect()[0]        # ONE Spark action

    columns = []
    for i, field in enumerate(fields):
        stat = {
            "name": field.name,
            "type": field.dataType.simpleString(),
            "null_or_blank_pct": round(100.0 * (row[f"_null_{i}"] or 0) / row_count, 2),
        }
        if f"_dist_{i}" in row.asDict():
            stat["approx_distinct"] = int(row[f"_dist_{i}"] or 0)
        columns.append(stat)

    return {"table_fqn": table_fqn, "row_count": row_count,
            "column_count": len(df.columns), "columns": columns}

Strip pipeline metadata before the prompt

Ingestion columns such as _ingested_at, _batch_id or whatever your platform adds will otherwise attract a dozen pointless rules. Filter them out of the pack, and say so in the prompt as well.

04Load your own documentation

This is the single biggest lever on rule quality, and it costs nothing but writing files.

Without context the model sees column names and null rates. It produces predictably generic output: null checks, length bounds, type validation. Those rules run, they pass, and they catch almost nothing.

With context it writes rules that encode what your business actually means. Valid status codes for this entity. That a lease end date must follow its start date. That a postcode format depends on the country column. Those are the rules that find real problems.

Use two tiers

The shared tier compounds. Write a domain standards document once, and every related table onboarded in future inherits it with no per-table work, because retrieval pulls only the passages matching that table’s column names. The same document serves each table differently.

src/dq/context.py

from pathlib import Path

READABLE = {".md", ".yaml", ".yml", ".txt", ".csv"}
PER_FILE_CHARS = 4_000

def build_context_bundle(root: Path, source_dir: str | None) -> dict:
    """Concatenate context files into one bundle with provenance headers."""
    parts, files = [], []
    dirs = [root / "data/context/default"]
    if source_dir:
        dirs.append(root / source_dir)

    for d in dirs:
        if not d.exists():
            continue
        for f in sorted(d.rglob("*")):
            if f.suffix.lower() not in READABLE:
                continue
            text = f.read_text(encoding="utf-8", errors="ignore")[:PER_FILE_CHARS]
            rel = f.relative_to(root).as_posix()
            parts.append(f"### {rel}\n{text}")     # header = provenance
            files.append(rel)

    return {"text_bundle": "\n\n".join(parts), "files": files,
            "has_source_context": bool(source_dir)}

The ### <path> headers matter. The chunker splits on them, so every chunk carries the file it came from, which means a retrieved passage can be traced back to its source document.

A CSV of code lists is a perfectly good context file. Capping each file at a few thousand characters embeds its first few dozen rows, which is usually enough for the valid values to be retrievable.

05Build the retrieval index

A JSONL file on a Volume, upserted by content hash so unchanged text is never re-embedded.

One JSON object per chunk:

{"content_hash": "a3f9c2b1d4e8f001",
 "text": "status_code valid values: ACTIVE, PENDING, CLOSED, VOID",
 "file_path": "data/context/default/code_lists.md",
 "chunk_idx": 2,
 "chunk_type": "context",
 "embedding": [0.0214, -0.0341, ...],
 "model": "databricks-gte-large-en"}

Chunking

Fixed word windows with overlap. The overlap exists so a definition that straddles a boundary survives whole in at least one chunk.

import hashlib

CHUNK_WORDS, OVERLAP = 250, 50

def chunk_bundle(text_bundle: str) -> list[dict]:
    chunks = []
    for section in text_bundle.split("### ")[1:]:
        path, _, body = section.partition("\n")
        words = body.split()
        step = CHUNK_WORDS - OVERLAP                  # 200
        for idx, start in enumerate(range(0, max(len(words), 1), step)):
            window = " ".join(words[start:start + CHUNK_WORDS])
            if not window.strip():
                continue
            chunks.append({
                "text": window,
                "file_path": path.strip(),
                "chunk_idx": idx,
                "chunk_type": "context",
                "content_hash": hashlib.sha256(window.encode()).hexdigest()[:16],
            })
    return chunks

With 250-word windows and 50-word overlap the step is 200, so chunk 0 is words 0 to 249, chunk 1 is words 200 to 449, and so on. A typical context folder produces between 80 and 200 chunks.

Calling the embedding endpoint

import requests

EMBED_ENDPOINT = "databricks-gte-large-en"
BATCH = 32

def embed_texts(texts: list[str]) -> list[list[float]]:
    ctx  = dbutils.notebook.entry_point.getDbutils().notebook().getContext()
    host = ctx.apiUrl().get()
    tok  = ctx.apiToken().get()

    out: list[list[float]] = []
    for i in range(0, len(texts), BATCH):
        batch = texts[i:i + BATCH]
        r = requests.post(
            f"{host}/serving-endpoints/{EMBED_ENDPOINT}/invocations",
            headers={"Authorization": f"Bearer {tok}",
                     "Content-Type": "application/json"},
            json={"input": batch}, timeout=120)
        r.raise_for_status()
        data = sorted(r.json()["data"], key=lambda d: d["index"])   # see note
        out.extend(d["embedding"] for d in data)
    return out

Upsert, do not rebuild

Embedding is the slow and billable step, and documentation changes rarely. Hash each chunk, keep the vectors you already have, and embed only what is new.

import json, os

def build_rag_index(text_bundle: str, volume_path: str,
                    index_file: str = "rag_index.jsonl") -> dict:
    path = os.path.join(volume_path, index_file)

    current = {c["content_hash"]: c for c in chunk_bundle(text_bundle)}

    existing = {}
    if os.path.exists(path):
        with open(path, encoding="utf-8") as fh:
            for line in fh:
                rec = json.loads(line)
                existing[rec["content_hash"]] = rec

    new_hashes = [h for h in current if h not in existing]
    kept       = {h: existing[h] for h in current if h in existing}
    dropped    = len(existing) - len(kept)          # deleted context files

    print(f"RAG upsert: {len(current)} current | {len(existing)} existing | "
          f"{len(new_hashes)} to embed | {dropped} dropped")

    records = dict(kept)
    if new_hashes:
        vectors = embed_texts([current[h]["text"] for h in new_hashes])
        for h, vec in zip(new_hashes, vectors):
            records[h] = {**current[h], "embedding": vec, "model": EMBED_ENDPOINT}

    with open(path, "w", encoding="utf-8") as fh:
        for rec in records.values():
            fh.write(json.dumps(rec) + "\n")

    return {"total": len(records), "embedded": len(new_hashes), "dropped": dropped}

Deleted context files need no special handling. Their hashes are in existing but not in current, so they are simply not written to the new file. If nothing changed, this makes zero API calls.

06Retrieve for this specific table

The query is built from what you already know about the table. You are asking which parts of your documentation describe this kind of data.

def build_query(application: str, entity: str, table_fqn: str,
                columns: list[str]) -> str:
    return (f"{application} {entity} {table_fqn} data quality "
            + " ".join(columns[:40]))

Column names are the strongest signal available. They tell the embedding model what the table is about, which is why a chunk describing valid status codes scores highly for a table that has a status_code column. Around forty columns is the useful limit; beyond that you add noise without improving retrieval.

import math

def cosine(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na  = math.sqrt(sum(x * x for x in a))
    nb  = math.sqrt(sum(y * y for y in b))
    return dot / (na * nb) if na and nb else 0.0

MAX_CONTEXT_CHARS = 120_000

def retrieve_context(query: str, volume_path: str, top_k: int = 12,
                     index_file: str = "rag_index.jsonl") -> str:
    path = os.path.join(volume_path, index_file)
    if not os.path.exists(path):
        return ""                                   # caller falls back

    with open(path, encoding="utf-8") as fh:
        index = [json.loads(line) for line in fh]

    q = embed_texts([query])[0]
    ranked = sorted(index, key=lambda r: cosine(q, r["embedding"]), reverse=True)

    picked = [f"### {r['file_path']}\n{r['text']}" for r in ranked[:top_k]]
    return "\n\n".join(picked)[:MAX_CONTEXT_CHARS]

Pure Python cosine similarity over a couple of hundred 1024-dimensional vectors runs in well under a second. You do not need numpy, and you certainly do not need a vector database, until the index grows by an order of magnitude.

07Generate the ruleset

The prompt is a specification. Every constraint you leave out is one the validator has to catch later.

Ask for the DQX check metadata shape directly, so there is no translation layer between what the model returns and what the engine executes.

src/dq/generate.py

SYSTEM_PROMPT = (
    "You are a data quality engineer. Output exactly one JSON object with keys "
    "`params` (object) and `datasets` (object). "
    "`datasets` must have one key (snake_case dataset name). Each dataset value has: "
    "`source` {`type`: `table`, `location`: must equal the `table` field in the user JSON}, "
    "and a `checks` array. "
    "Each check: `rule_id` (int), `rule_description`, `dimension_description`, "
    "`criticality` (`error` or `warn`), nested `check`: "
    "{`function`: `sql_expression`, `arguments`: {`expression`}}. "
    "Use ONLY column names from table_description_pack.columns[].name (exact spelling). "
    "Do not invent columns (e.g. customer_status if the table has status). "
    "Quote columns containing / or spaces with backticks. "
    "Use try_cast (not cast) for numeric or date conversion; "
    "wrap with trim(cast(col as string)) first. "
    "sql_expression MUST evaluate to TRUE for valid rows; FALSE means violation. "
    "Example: `address_1 IS NOT NULL AND trim(address_1) != ''`. "
    "Never put the rule description text in expression. No English sentences. "
    "Every expression MUST reference at least one real column. "
    "No joins, no subqueries, no dotted paths. "
    "Use metadata_text when it is present. "
    "Cap checks at max_rules. No markdown. No text outside JSON."
)

Six of those lines exist because a model broke each of them in testing. The polarity line is the most important one in the file.

The user message

user_payload = {
    "application": source["application"],
    "entity": source["entity"],
    "table": source["table"]["location"],
    "max_rules": source["llm"]["max_rules"],
    "has_context": bool(metadata_text),
    "table_description_pack": table_pack,     # stage 1
    "metadata_text": metadata_text,           # stage 6, top-k chunks
}
user_prompt = json.dumps(user_payload, indent=2)[:180_000]

Calling the model

LangChain’s ChatOpenAI works against any OpenAI-compatible endpoint. Note that response_format is not a first-class constructor argument, so it goes through model_kwargs.

import os, json, re
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser

def make_llm(temperature: float = 0.1, max_tokens: int = 8000,
             json_mode: bool = True, timeout: int = 180) -> ChatOpenAI:
    kwargs = {"response_format": {"type": "json_object"}} if json_mode else {}
    return ChatOpenAI(
        model=os.environ["LLM_MODEL"],
        api_key=os.environ["LLM_API_KEY"],
        base_url=os.environ["LLM_BASE_URL"],
        temperature=temperature,
        max_tokens=max_tokens,
        timeout=timeout,
        model_kwargs=kwargs,
    )

def call_chat(system_prompt: str, user_prompt: str, **kw) -> str:
    chain = make_llm(**kw) | StrOutputParser()
    return chain.invoke([SystemMessage(content=system_prompt),
                         HumanMessage(content=user_prompt)])

Parse defensively

Models wrap JSON in markdown fences, or add a sentence of preamble. Try the raw string, then the inside of a fenced block, then the outermost brace-to-brace slice.

def parse_json_content(text: str) -> dict | None:
    raw = (text or "").strip()
    if not raw:
        return None

    candidates = [raw]
    fenced = re.search(r"```(?:json)?\s*([\s\S]*?)\s*```", raw, flags=re.IGNORECASE)
    if fenced:
        candidates.insert(0, fenced.group(1).strip())
    start, end = raw.find("{"), raw.rfind("}")
    if start >= 0 and end > start:
        candidates.append(raw[start:end + 1])

    for c in candidates:
        try:
            parsed = json.loads(c)
            if isinstance(parsed, dict):
                return parsed
        except json.JSONDecodeError:
            continue
    return None

The retry ladder

def generate_contract(system_prompt: str, user_prompt: str) -> dict:
    # 1. JSON mode
    try:
        out = parse_json_content(call_chat(system_prompt, user_prompt))
        if out:
            return out
    except Exception as ex:
        print(f"json-mode call failed, retrying without response_format: {ex}")
        # 2. some gateways reject response_format entirely
        out = parse_json_content(call_chat(system_prompt, user_prompt, json_mode=False))
        if out:
            return out

    # 3. blunter prompt, temperature 0
    strict = system_prompt + (
        "\n\nCRITICAL: Reply with a single valid JSON object only. "
        "No markdown fences, no commentary.")
    out = parse_json_content(call_chat(strict, user_prompt, temperature=0.0))
    if out:
        return out

    # 4. static baseline checked into the repo
    return load_fallback_contract()

08Validate and repair before anything runs

This is the stage that separates a demo from a pipeline. Treat generated SQL exactly as you would treat SQL submitted by an anonymous user.

src/dq/repair.py

PROSE = (" must ", " should ", " cannot be ", " not be null", " is required",
         " valid value", " each row", " must not ")

def looks_like_prose(expr: str) -> bool:
    return any(p in f" {(expr or '').lower()} " for p in PROSE)

SQL_WORDS = {"and","or","not","null","is","in","like","rlike","between","case",
             "when","then","else","end","cast","try_cast","trim","length","upper"}

def identifiers(expr: str) -> list[str]:
    found = re.findall(r"`([^`]+)`|\b([A-Za-z_][A-Za-z0-9_]*)\b", expr or "")
    names = [a or b for a, b in found]
    return [n for n in names if n.lower() not in SQL_WORDS]

def references_real_column(expr: str, columns: list[str]) -> bool:
    known = {c.lower() for c in columns}
    return any(i.lower() in known for i in identifiers(expr))

The probe:

def probe(df, expr: str) -> str | None:
    """Return None if the expression runs, else the first line of the error."""
    try:
        df.selectExpr(f"({expr}) AS _dq_probe_ok").limit(1).collect()
        return None
    except Exception as ex:
        return str(ex).splitlines()[0]

The template library. Unglamorous, and it recovers a meaningful share of otherwise dead rules:

def compile_from_description(desc: str, column: str) -> str | None:
    q = f"`{column}`" if re.search(r"[^\w]", column) else column
    s = f"trim(cast({q} as string))"
    d = (desc or "").lower()

    if "not be null" in d or "not be blank" in d or "is required" in d:
        return f"({q} IS NOT NULL AND {s} <> '')"
    if "must be numeric" in d or "must be a number" in d:
        return f"({q} IS NULL OR try_cast({s} as double) IS NOT NULL)"
    if "must be positive" in d or "greater than 0" in d:
        return f"({q} IS NULL OR try_cast({s} as double) > 0)"
    if "valid date" in d or "valid timestamp" in d:
        return f"({q} IS NULL OR try_to_timestamp({s}) IS NOT NULL)"
    if "yes or no" in d:
        return f"({q} IS NULL OR upper({s}) IN ('YES','NO','Y','N'))"
    if "valid email" in d:
        return f"({q} IS NULL OR ({s} LIKE '%@%.%' AND {s} NOT LIKE '%@%@%'))"
    if "latitude" in d:
        return f"({q} IS NULL OR try_cast({s} as double) BETWEEN -90 AND 90)"
    if "longitude" in d:
        return f"({q} IS NULL OR try_cast({s} as double) BETWEEN -180 AND 180)"
    return None

And the ladder that ties them together:

def ensure_runnable(df, expr: str, desc: str, columns: list[str]):
    """Return (safe_expression, skip_reason). One of the two is always None."""

    # 1. static
    if expr and not looks_like_prose(expr) and references_real_column(expr, columns):
        err = probe(df, expr)                       # 2. probe
        if err is None:
            return expr, None
    else:
        err = "failed static validation"

    # 3. template repair
    col = next((c for c in columns if c.lower() in (desc or "").lower()), None)
    if col:
        compiled = compile_from_description(desc, col)
        if compiled and probe(df, compiled) is None:
            return compiled, None

    # 4. LLM repair
    fixed = llm_repair(expr, desc, columns, err)
    if fixed and probe(df, fixed) is None:
        return fixed, None

    return None, err                                 # skip, never execute

def llm_repair(expr: str, desc: str, columns: list[str], err: str) -> str | None:
    system = ("You are a Spark SQL expert. Output ONLY the corrected boolean "
              "expression. Single line. No markdown, no explanation.")
    user = (f"Use ONLY these columns: {', '.join(columns[:60])}\n"
            f"Rule intent: {desc}\n"
            f"Broken expression: {expr}\n"
            f"Spark error: {err}")
    try:
        out = call_chat(system, user, temperature=0.0,
                        max_tokens=200, json_mode=False)
        return out.strip().strip("`").replace("```sql", "").replace("```", "").strip()
    except Exception:
        return None

09Execute with DQX

Because the model was asked for DQX check metadata, surviving rules go straight to the engine with no translation.

src/dq/execute.py

from databricks.labs.dqx.engine import DQEngine
from databricks.sdk import WorkspaceClient
from pyspark.sql import functions as F

_engine = None

def get_engine() -> DQEngine:
    global _engine
    if _engine is None:
        _engine = DQEngine(WorkspaceClient())
    return _engine

def run_check(df, check: dict, total: int) -> dict:
    """Execute one generated check and return pass/fail counts."""
    dq_check = {
        "name": f"rule_{check['rule_id']}",
        "criticality": check.get("criticality", "error"),
        "check": check["check"],                      # already DQX shape
        "user_metadata": {"rule_id": str(check["rule_id"]),
                          "dimension": str(check.get("dimension_description", ""))},
    }
    out = get_engine().apply_checks_by_metadata(df, [dq_check])

    fail = out.filter(F.col("_errors").isNotNull()).count()
    return {"rule_id": check["rule_id"],
            "rule_description": check["rule_description"],
            "dimension": check.get("dimension_description"),
            "expression": check["check"]["arguments"].get("expression"),
            "check_fail_count": fail,
            "check_pass_count": total - fail,
            "total_count": total}

DQX annotates the DataFrame with _errors and _warnings columns rather than returning counts, so you derive the numbers yourself. If you want the split instead of the annotation, apply_checks_by_metadata_and_split returns valid and quarantined DataFrames.

10Score the result

Two scores, measuring different things. Keep them apart.

Profile score: how healthy is the table

Computed from the profile, independent of any rule. Four components summing to 100:

def profile_score(stats, pk_dup_pct, has_recent_commit, test_pct):
    avg_null  = sum(s["null_or_blank_pct"] for s in stats) / max(len(stats), 1)
    high_null = sum(1 for s in stats if s["null_or_blank_pct"] > 10.0)

    completeness = max(0.0, 50.0 - avg_null * 1.8 - high_null * 1.5)
    uniqueness   = max(0.0, 25.0 * (1.0 - pk_dup_pct / 100.0)) if pk_dup_pct >= 0 else 18.0
    freshness    = 15.0 if has_recent_commit else 5.0
    test_value   = max(0.0, 10.0 - test_pct * 2.0)

    return round(min(100.0, completeness + uniqueness + freshness + test_value))

Freshness comes from DESCRIBE HISTORY <table> LIMIT 1, which is free on any Delta table. Duplicate rate comes from the configured primary key:

dups = (df.groupBy(*[F.col(f"`{c}`") for c in pk_cols]).count()
          .filter(F.col("count") > 1)
          .agg(F.sum(F.col("count") - 1).alias("dup_rows"))
          .collect()[0]["dup_rows"]) or 0
pk_dup_pct = 100.0 * dups / row_count

DQ score: how many rules passed

The version the report badge shows is simply the share of executed rules that had zero exceptions:

def dq_score(results) -> int:
    executed = len(results)
    if not executed:
        return 0
    passed = sum(1 for r in results if r["check_fail_count"] == 0)
    return round(100 * passed / executed)

The denominator is rules executed, not rules generated. Rules the repair ladder skipped are excluded, which is right, because a rule that never ran is not evidence of anything. Record both counts separately so that a shrinking denominator cannot quietly inflate the score.

Weighting it, once the simple version stops being useful

A flat pass rate treats a null primary key and a formatting slip identically. Once your rules carry dimensions, weight the penalty by violation rate instead:

WEIGHTS = {"completeness": 3.0, "uniqueness": 3.0, "validity": 2.0,
           "consistency": 2.0, "accuracy": 1.5, "format": 1.0}

def weighted_dq_score(results, weights=WEIGHTS) -> float:
    if not results:
        return 100.0
    penalty = sum(weights.get(r["dimension"], 1.0)
                  * (r["check_fail_count"] / max(r["total_count"], 1)) * 100.0
                  for r in results)
    return max(0.0, 100.0 - min(penalty, 100.0))

This one is rate-based rather than rule-based. A rule failing on 0.1% of rows barely moves it; the same rule failing on 40% dominates. The weights are a policy decision, not a technical one, so put them somewhere your data stewards can argue with them.

11The report

The job’s output is a single interactive HTML report per table, rendered straight into the notebook. Three parts, in this order.

Part 1: written summary and data profile

Above the table sits a short written summary, then the ten worst columns by null rate with a traffic-light signal, then the structural heuristics (blank strings, whitespace padding, constant columns, all-null columns, date range sentinels, cardinality outliers, key duplication, test values), then the profile score badge. Print the formula with its live values next to the score. A bare number invites argument; C=0 + P=25 + V=15 + T=10 ends it, because the reader can see that completeness scored zero and why.

Part 2: quality results and analytics

Four summary cards, each clickable to expand the underlying rule list inline. Note that they answer two different questions: rule-level (how many rules found anything) and record-level (how many rows are clean against every rule). The second number is usually far worse than people expect the first time they see it, because a row needs to pass all fifty rules to count as clean.

Then the analytics panel:

The Pareto panel earns its place more than the rest. On a first run it almost always shows that three or four rules produce nearly all exceptions, which turns an intimidating “28 rules failing” into a morning’s work.

Part 3: the rule table

Every rule evaluated as a plain DataFrame: description, dimension, pass count, fail count, total rows, the validated SQL expression that actually ran, and any error. This is the part people export and send to a data owner, so it matters that the expression column shows the repaired SQL rather than what the model originally produced.

Building it: no charting library

Build HTML strings in Python and hand them to displayHTML.

This sounds primitive and is the right call. Databricks renders displayHTML output in a sandboxed iframe, so external CSS and JS files do not load. Inline style= attributes and inline SVG always work.

src/dq/report.py

import html as html_module
import sys

esc = html_module.escape        # escape every data-derived string

def show_html(markup: str) -> None:
    main = sys.modules.get("__main__")
    fn = getattr(main, "displayHTML", None) if main else None
    if callable(fn):
        fn(markup)
        return
    try:                                  # plain Jupyter fallback
        from IPython.display import HTML, display
        display(HTML(markup))
    except Exception:
        print("(cannot render HTML here)")

cells = [donut_html, gauge_html, dimension_bars_html, treemap_html,
         top_failures_html, pareto_html, histogram_html, rule_explorer_html]

show_html(
    "<div style='background:#0f172a;padding:16px;border-radius:8px'>"
    "<div style='display:grid;grid-template-columns:1fr 1fr;gap:14px'>"
    + "".join(c for c in cells if c) +
    "</div></div>"
)                                          # ONE call, one iframe

What to draw, and how:

A donut is about fifteen lines of SVG. Circumference is 2πr, each segment sets stroke-dasharray to its share of that, and stroke-dashoffset accumulates as you go.

A written summary

One more model call turns the numbers into something a steward reads. Send the aggregates only, never rows.

def summarise(table_pack: dict, results: list[dict]) -> str:
    payload = {
        "table": table_pack["table_fqn"],
        "row_count": table_pack["row_count"],
        "column_count": table_pack["column_count"],
        "worst_null_columns": sorted(table_pack["columns"],
            key=lambda c: -c["null_or_blank_pct"])[:15],
        "failing_rules": [r for r in results if r["check_fail_count"] > 0][:15],
    }
    system = ("You are a data analyst writing for data stewards. Write 3 to 5 "
              "sentences in plain English: what this table contains, the main "
              "quality issues, and what to prioritise. Do not list every column.")
    try:
        return call_chat(system, json.dumps(payload)[:12_000],
                         temperature=0.2, max_tokens=400, json_mode=False)
    except Exception:
        return deterministic_summary(payload)      # template fallback

Keep the deterministic fallback. A report that fails to render because a summary call timed out is a bad trade.

12Persist, then trend

A score you look at once is a curiosity. The value is in the series.

CREATE TABLE IF NOT EXISTS main.dq.scorecard_run (
  run_id            STRING,
  run_ts            TIMESTAMP,
  source_id         STRING,
  table_name        STRING,
  profile_score     DOUBLE,
  dq_score          DOUBLE,
  rows_total        BIGINT,
  rules_generated   INT,
  rules_executed    INT,      -- after the repair ladder
  ruleset_hash      STRING,   -- attribute score moves to rules vs data
  generation_mode   STRING    -- 'llm' or 'fallback'
) USING DELTA;

CREATE TABLE IF NOT EXISTS main.dq.scorecard_rule (
  run_id            STRING,
  rule_id           INT,
  rule_description  STRING,
  dimension         STRING,
  expression        STRING,
  pass_count        BIGINT,
  fail_count        BIGINT,
  total_count       BIGINT
) USING DELTA;

Write with the DataFrame writer, not string-concatenated INSERT statements. Rule descriptions contain apostrophes, and hand-built SQL is how you get an injection bug in your own pipeline.

spark.createDataFrame(rule_rows).write.mode("append").saveAsTable("main.dq.scorecard_rule")

Storing ruleset_hash is what makes the trend line trustworthy. A score that moves while the hash is unchanged is a data change. A score that moves with a new hash is uninvestigated until proven otherwise.

The flywheel

Once rules are executing, feed the good ones back into the retrieval index as a second chunk type. Embed the rule description, its dimension, its SQL and its result, and append it.

def append_exemplar(rule: dict, volume_path: str) -> None:
    text = (f"Rule: {rule['rule_description']}\n"
            f"Dimension: {rule['dimension']}\n"
            f"SQL: {rule['expression']}\n"
            f"Result: {rule['check_fail_count']}/{rule['total_count']} rows failed")
    vec = embed_texts([text])[0]
    rec = {"content_hash": hashlib.sha256(text.encode()).hexdigest()[:16],
           "text": text, "file_path": "exemplars", "chunk_idx": 0,
           "chunk_type": "exemplar_rule", "embedding": vec, "model": EMBED_ENDPOINT}
    with open(os.path.join(volume_path, "rag_index.jsonl"), "a", encoding="utf-8") as fh:
        fh.write(json.dumps(rec) + "\n")

On the next run, a table with similar columns retrieves those proven rules as context. The model gets working examples from your own warehouse rather than inventing from scratch, and rule quality compounds without anyone retraining anything. Deduplicate by content hash so reruns do not bloat the index.

13Wire the notebook

Three cells.

notebooks/run_dq.py

# ---- Cell 1: setup -------------------------------------------------------
%pip install databricks-labs-dqx==0.15.0 langchain-openai pyyaml
dbutils.library.restartPython()

# ---- Cell 2: config, context, index --------------------------------------
import yaml, sys
from pathlib import Path

ROOT = Path("/Workspace/Repos/your.name/dq_pipeline")
sys.path.insert(0, str(ROOT / "src"))

from dq.context import build_context_bundle
from dq.rag import build_rag_index

cfg = yaml.safe_load((ROOT / "configs/dq_config.yaml").read_text())

bundle = build_context_bundle(ROOT, source_dir=None)      # default tier
stats  = build_rag_index(bundle["text_bundle"], cfg["rag"]["volume_path"])
print(stats)   # {'total': 143, 'embedded': 5, 'dropped': 2}

# ---- Cell 3: run every source -------------------------------------------
from dq.runner import run_source
from dq.report import render

for source in cfg["sources"]:
    result = run_source(spark, source, cfg)
    render(result)

And run_source is just the stages in order:

def run_source(spark, source, cfg) -> dict:
    table = source["table"]["location"]
    df    = spark.read.table(table)
    total = df.count()

    pack    = build_table_pack(df, table, total)                       # 1
    bundle  = build_context_bundle(ROOT, source.get("context_directory"))  # 2
    query   = build_query(source["application"], source["entity"], table,
                          [c["name"] for c in pack["columns"]])
    context = (retrieve_context(query, cfg["rag"]["volume_path"],
                                cfg["rag"]["top_k"])
               or bundle["text_bundle"][:120_000])                     # 4 + fallback

    contract = generate_contract(SYSTEM_PROMPT,
                                 build_user_prompt(source, pack, context))  # 5
    checks   = extract_checks(contract)[: source["llm"]["max_rules"]]

    results, skipped = [], []
    columns = [c["name"] for c in pack["columns"]]
    for check in checks:
        expr = check["check"]["arguments"].get("expression", "")
        safe, reason = ensure_runnable(df, expr, check["rule_description"], columns)  # 6
        if safe is None:
            skipped.append({**check, "skip_reason": reason})
            continue
        check["check"]["arguments"]["expression"] = safe
        results.append(run_check(df, check, total))                    # 7

    return {"table_pack": pack, "results": results, "skipped": skipped,
            "dq_score": dq_score(results),                             # 8
            "rules_generated": len(checks), "rules_executed": len(results)}

14Failure modes and what happens

The common thread: every stage degrades rather than throwing. A broken rule is skipped and logged, not executed. A missing index costs quality, not availability. The only thing the pipeline refuses to do is run SQL it has not proven it can run.

What never goes to the model

• Row data. The model sees column names, types, null percentages, distinct counts and retrieved documentation. It never sees a value from the table.
• Credentials or tokens.
• Pipeline metadata columns, filtered out before the prompt is built.

That boundary is worth writing down before your first governance review, because it is the question you will be asked.

The generation is the easy part. The system is the profile, the retrieval, the repair ladder and the trend line.

---