How should Microsoft Fabric data pipelines handle AI-driven validation ?

Hi codeautomation,

The architecture above is the right shape, so I'll add a concrete notebook version of it. It covers your points in one place: schema checks before the write, fail vs warn, logging, retries and keeping it cheap on large data.

1. Contract check before anything else, so schema drift fails fast with a clear message instead of surfacing later in the semantic model:

from pyspark.sql import functions as F
import json, uuid

run_id = str(uuid.uuid4())
df = spark.read.table("bronze.orders")

expected = {"order_id": "int", "customer_id": "int",
            "amount": "decimal(18,2)", "order_date": "date"}
actual = dict(df.dtypes)
drift = {c: (t, actual.get(c)) for c, t in expected.items()
         if actual.get(c) != t}

if drift:
    notebookutils.notebook.exit(json.dumps(
        {"status": "FAIL", "reason": "schema",
         "detail": str(drift), "run_id": run_id}))

2. Deterministic rules with a severity each, evaluated in one pass with Spark:

rules = {
    "null_key":    ("critical", F.col("order_id").isNull()),
    "neg_amount":  ("warning",  F.col("amount") < 0),
    "future_date": ("warning",  F.col("order_date") > F.current_date()),
}

flagged = df
for name, (_, cond) in rules.items():
    flagged = flagged.withColumn(
        name, F.coalesce(cond, F.lit(False)))
flagged = flagged.cache()

counts = flagged.agg(*[
    F.sum(F.col(n).cast("int")).alias(n) for n in rules
]).first().asDict()

dup_keys = (df.groupBy("order_id").count()
              .filter("count > 1").count())

critical = (counts["null_key"] or 0) + dup_keys
warnings = sum(counts[n] or 0 for n, (s, _) in rules.items()
               if s == "warning")

3. Log every rule, every run, then quarantine instead of silently fixing:

log = [(run_id, "orders", n, s, int(counts[n] or 0))
       for n, (s, _) in rules.items()]
log.append((run_id, "orders", "duplicate_key", "critical", dup_keys))

(spark.createDataFrame(
    log, "run_id string, tbl string, rule string, "
         "severity string, failed_rows long")
  .withColumn("run_ts", F.current_timestamp())
  .write.mode("append").saveAsTable("dq.validation_log"))

any_issue = " OR ".join(rules)
(flagged.filter(any_issue)
  .withColumn("run_id", F.lit(run_id))
  .write.mode("append").saveAsTable("dq.orders_quarantine"))

if critical == 0:
    (flagged.filter(f"NOT ({any_issue})").drop(*rules)
      .write.mode("append").saveAsTable("silver.orders"))

4. AI only on what the rules could not decide, and only once per row. Hashing the row content means retries and reruns never pay twice for the same record:

suspects = (flagged.filter("neg_amount OR future_date")
  .withColumn("row_hash",
      F.sha2(F.to_json(F.struct(*df.columns)), 256)))

reviewed = spark.read.table("dq.ai_review_cache").select("row_hash")
to_review = suspects.join(reviewed, "row_hash", "left_anti")

# run Fabric AI functions (classify / explain) on to_review only,
# then append row_hash + verdict to dq.ai_review_cache

5. Return one result to the pipeline:

status = "FAIL" if critical else ("WARN" if warnings else "PASS")
notebookutils.notebook.exit(json.dumps({
    "status": status, "run_id": run_id,
    "critical": critical, "warnings": warnings}))

In the pipeline, the If Condition reads the exit value:

@equals(
  json(activity('Validate').output.result.exitValue).status,
  'FAIL')

For retries, I would set a retry policy on the notebook activity for technical errors (timeouts, throttling) only. A data-quality FAIL should never be retried, because the same data fails the same way.

So my overall approach would be:

contract check -> fail fast on schema drift
-> Spark rules with severity
-> log every rule, quarantine failed rows
-> AI only on unresolved rows, cached by hash
-> exit value drives pipeline branching

The dq.validation_log table also gives you a simple Power BI view of data quality trends per table and rule over time.

AI-assisted drafting: AI was used to help structure and phrase this response. I reviewed and validated the technical content before posting.