Building an AI-Powered Data Quality Pipeline: Automated Validation, Anomaly Detection, and Schema Drift Prevention

At three data sources, a list of hand-written validation rules is manageable. At thirty, it is a maintenance burden. At three hundred, it is a lie. You have rules that were accurate eighteen months ago, rules that were never updated after the upstream team changed their enum values, and columns that no rule covers because nobody remembered they existed.

The failure mode is not dramatic. Bad data arrives, passes validation, flows into downstream aggregations, and surfaces two weeks later as a corrupted dashboard or a model trained on garbage. The investigation traces back to a column that changed type silently and a rule that should have caught it but did not.

This article covers how to replace the hand-written rule list with a pipeline that profiles data statistically, detects distribution anomalies on each ingestion batch, identifies schema drift using embedding similarity, enforces data contracts generated from documentation, and quarantines bad data before it reaches production systems.

---

Why Rule-Based Validation Fails at Scale

Rule-based validation has a well-understood lifecycle. A data engineer writes rules for a new source: not-null checks, value range assertions, referential integrity constraints. Those rules are accurate on the day they are written. Over time, upstream systems change. A string column starts arriving as an integer. A field that was always populated starts coming through with 30% nulls. An enum grows new values that the rule’s allowlist does not include.

The rules do not update automatically. Nobody has a complete list of which upstream sources feed which validation rules. The engineer who wrote the rules left six months ago.

Statistical and ML-based approaches do not replace rules entirely. They augment rules with learned baselines so that changes that deviate meaningfully from historical patterns get flagged, even when no explicit rule covers them.

The shift is from “does this column satisfy a predicate” to “does this column’s current distribution match what we have seen before.”

---

Statistical Profiling and Baseline Generation

The first step is generating a statistical profile for each column across historical batches. This profile becomes the baseline against which incoming data is evaluated.

interface ColumnProfile {
  columnName: string;
  dataType: "numeric" | "categorical" | "timestamp" | "text";
  nullRate: number;
  distinctCount: number;
  // Numeric columns
  mean?: number;
  stdDev?: number;
  p5?: number;
  p25?: number;
  p50?: number;
  p75?: number;
  p95?: number;
  // Categorical columns
  topValues?: Record<string, number>; // value -> frequency
  cardinalityRatio?: number; // distinctCount / rowCount
}

interface DatasetProfile {
  sourceId: string;
  tableName: string;
  rowCount: number;
  batchTimestamp: string;
  columns: ColumnProfile[];
  profileVersion: number;
}

async function profileDataset(
  rows: Record<string, unknown>[],
  sourceId: string,
  tableName: string
): Promise<DatasetProfile> {
  const columns = new Map<string, unknown[]>();

  for (const row of rows) {
    for (const [key, value] of Object.entries(row)) {
      if (!columns.has(key)) columns.set(key, []);
      columns.get(key)!.push(value);
    }
  }

  const profiles: ColumnProfile[] = [];

  for (const [columnName, values] of columns) {
    const nullCount = values.filter(
      (v) => v === null || v === undefined
    ).length;
    const nonNullValues = values.filter((v) => v !== null && v !== undefined);
    const distinct = new Set(nonNullValues.map(String));

    const inferred = inferType(nonNullValues);

    const base: ColumnProfile = {
      columnName,
      dataType: inferred,
      nullRate: nullCount / values.length,
      distinctCount: distinct.size,
    };

    if (inferred === "numeric") {
      const nums = nonNullValues.map(Number).sort((a, b) => a - b);
      const mean = nums.reduce((s, n) => s + n, 0) / nums.length;
      const variance =
        nums.reduce((s, n) => s + (n - mean) ** 2, 0) / nums.length;

      Object.assign(base, {
        mean,
        stdDev: Math.sqrt(variance),
        p5: percentile(nums, 5),
        p25: percentile(nums, 25),
        p50: percentile(nums, 50),
        p75: percentile(nums, 75),
        p95: percentile(nums, 95),
      });
    } else if (inferred === "categorical") {
      const freq: Record<string, number> = {};
      for (const v of nonNullValues) {
        const s = String(v);
        freq[s] = (freq[s] ?? 0) + 1;
      }
      const top = Object.entries(freq)
        .sort((a, b) => b[1] - a[1])
        .slice(0, 20)
        .reduce<Record<string, number>>((acc, [k, v]) => {
          acc[k] = v / nonNullValues.length;
          return acc;
        }, {});

      Object.assign(base, {
        topValues: top,
        cardinalityRatio: distinct.size / values.length,
      });
    }

    profiles.push(base);
  }

  return {
    sourceId,
    tableName,
    rowCount: rows.length,
    batchTimestamp: new Date().toISOString(),
    columns: profiles,
    profileVersion: 1,
  };
}

Profiles are stored per batch and aggregated into a rolling baseline over the last N batches (typically 30 days or 90 ingestion runs, whichever is larger).

---

Anomaly Scoring on Incoming Batches

Given a baseline profile and an incoming batch profile, the anomaly scorer computes a score per column. Columns with scores above a threshold generate alerts or route data to quarantine.

Z-Score Scoring for Numeric Columns

For numeric columns, the key signals are: null rate shift, mean shift, and distribution shape change (measured via percentile drift).

interface AnomalySignal {
  columnName: string;
  signalType:
    | "null_rate_spike"
    | "mean_shift"
    | "distribution_shift"
    | "cardinality_explosion"
    | "new_values"
    | "type_change";
  severity: "low" | "medium" | "high" | "critical";
  zScore?: number;
  baseline: unknown;
  observed: unknown;
  description: string;
}

function scoreNumericColumn(
  baseline: ColumnProfile,
  incoming: ColumnProfile,
  baselineStdDev: {
    nullRate: number;
    mean: number;
    p50: number;
  }
): AnomalySignal[] {
  const signals: AnomalySignal[] = [];

  // Null rate change
  const nullZScore =
    Math.abs(incoming.nullRate - baseline.nullRate) /
    (baselineStdDev.nullRate + 0.001);

  if (nullZScore > 3) {
    signals.push({
      columnName: incoming.columnName,
      signalType: "null_rate_spike",
      severity: nullZScore > 6 ? "critical" : "high",
      zScore: nullZScore,
      baseline: baseline.nullRate,
      observed: incoming.nullRate,
      description: `Null rate changed from ${(baseline.nullRate * 100).toFixed(1)}% to ${(incoming.nullRate * 100).toFixed(1)}%`,
    });
  }

  // Mean shift
  if (baseline.mean !== undefined && incoming.mean !== undefined) {
    const meanZScore =
      Math.abs(incoming.mean - baseline.mean) /
      (baselineStdDev.mean + 0.001);

    if (meanZScore > 3) {
      signals.push({
        columnName: incoming.columnName,
        signalType: "mean_shift",
        severity: meanZScore > 5 ? "high" : "medium",
        zScore: meanZScore,
        baseline: baseline.mean,
        observed: incoming.mean,
        description: `Mean shifted from ${baseline.mean.toFixed(2)} to ${incoming.mean.toFixed(2)}`,
      });
    }
  }

  return signals;
}

Isolation Forest for Batch-Level Anomaly Scoring

Z-scores work column by column but miss correlated anomalies where no single column looks wrong but the batch as a whole is unusual. An isolation forest operates on the full feature vector of a batch profile.

interface BatchFeatureVector {
  batchId: string;
  features: number[]; // flattened profile metrics
  featureNames: string[];
}

function profileToFeatureVector(profile: DatasetProfile): BatchFeatureVector {
  const features: number[] = [];
  const featureNames: string[] = [];

  features.push(profile.rowCount);
  featureNames.push("row_count");

  for (const col of profile.columns) {
    features.push(col.nullRate);
    featureNames.push(`${col.columnName}.null_rate`);

    features.push(col.distinctCount);
    featureNames.push(`${col.columnName}.distinct_count`);

    if (col.mean !== undefined) {
      features.push(col.mean);
      featureNames.push(`${col.columnName}.mean`);
    }

    if (col.stdDev !== undefined) {
      features.push(col.stdDev);
      featureNames.push(`${col.columnName}.std_dev`);
    }

    if (col.cardinalityRatio !== undefined) {
      features.push(col.cardinalityRatio);
      featureNames.push(`${col.columnName}.cardinality_ratio`);
    }
  }

  return {
    batchId: profile.batchTimestamp,
    features,
    featureNames,
  };
}

// Simplified isolation tree node
interface IsolationNode {
  isLeaf: boolean;
  splitFeatureIndex?: number;
  splitValue?: number;
  left?: IsolationNode;
  right?: IsolationNode;
  depth: number;
  size: number;
}

function anomalyScore(pathLength: number, n: number): number {
  // Average path length of unsuccessful search in BST
  const c = n <= 1 ? 1 : 2 * (Math.log(n - 1) + 0.5772156649) - (2 * (n - 1)) / n;
  return Math.pow(2, -pathLength / c);
}

In practice, you would use a mature isolation forest implementation (Python’s scikit-learn via a subprocess, or a WASM port). The pattern above shows how to construct the feature vector from profiles so the model operates on summary statistics rather than raw row data, which is critical for performance at scale.

---

Schema Drift Detection with Embedding Similarity

Schema drift is subtler than column deletion. A column gets renamed from customer_id to customerId. A field that encoded status as integers now encodes it as strings. A nested object gets flattened.

Embedding-based schema comparison catches these cases by representing column names and their value distributions as vectors and measuring cosine similarity between schema versions.

interface ColumnEmbedding {
  columnName: string;
  nameEmbedding: number[];
  distributionEmbedding: number[]; // embedding of value samples or top-N values
  combined: number[];
}

async function embedColumn(
  col: ColumnProfile,
  sampleValues: string[],
  embed: (text: string) => Promise<number[]>
): Promise<ColumnEmbedding> {
  const nameEmbedding = await embed(col.columnName);

  // For categorical columns, embed the top value list as context
  const distributionContext =
    col.topValues
      ? Object.keys(col.topValues).slice(0, 10).join(", ")
      : sampleValues.slice(0, 10).join(", ");

  const distributionEmbedding = await embed(
    `${col.columnName}: ${distributionContext}`
  );

  // Combine name and distribution embeddings with 60/40 weighting
  const combined = nameEmbedding.map(
    (v, i) => v * 0.6 + distributionEmbedding[i] * 0.4
  );

  return {
    columnName: col.columnName,
    nameEmbedding,
    distributionEmbedding,
    combined,
  };
}

function cosineSimilarity(a: number[], b: number[]): number {
  const dot = a.reduce((sum, val, i) => sum + val * b[i], 0);
  const magA = Math.sqrt(a.reduce((sum, val) => sum + val * val, 0));
  const magB = Math.sqrt(b.reduce((sum, val) => sum + val * val, 0));
  return dot / (magA * magB + 1e-8);
}

interface SchemaDriftResult {
  matchedColumns: Array<{
    baseline: string;
    incoming: string;
    similarity: number;
    driftType: "exact" | "renamed" | "semantically_shifted" | "unmatched";
  }>;
  newColumns: string[];
  removedColumns: string[];
  overallDriftScore: number;
}

function detectSchemaDrift(
  baselineEmbeddings: ColumnEmbedding[],
  incomingEmbeddings: ColumnEmbedding[],
  renameThreshold = 0.88,
  shiftThreshold = 0.72
): SchemaDriftResult {
  const matched = new Set<string>();
  const results: SchemaDriftResult["matchedColumns"] = [];

  for (const incoming of incomingEmbeddings) {
    let bestMatch: { col: ColumnEmbedding; sim: number } | null = null;

    for (const baseline of baselineEmbeddings) {
      const sim = cosineSimilarity(incoming.combined, baseline.combined);
      if (!bestMatch || sim > bestMatch.sim) {
        bestMatch = { col: baseline, sim };
      }
    }

    if (!bestMatch || bestMatch.sim < shiftThreshold) {
      results.push({
        baseline: "",
        incoming: incoming.columnName,
        similarity: bestMatch?.sim ?? 0,
        driftType: "unmatched",
      });
      continue;
    }

    matched.add(bestMatch.col.columnName);
    const isExact = bestMatch.col.columnName === incoming.columnName;
    const driftType = isExact
      ? "exact"
      : bestMatch.sim >= renameThreshold
      ? "renamed"
      : "semantically_shifted";

    results.push({
      baseline: bestMatch.col.columnName,
      incoming: incoming.columnName,
      similarity: bestMatch.sim,
      driftType,
    });
  }

  const removedColumns = baselineEmbeddings
    .filter((b) => !matched.has(b.columnName))
    .map((b) => b.columnName);

  const newColumns = results
    .filter((r) => r.driftType === "unmatched")
    .map((r) => r.incoming);

  const avgSimilarity =
    results
      .filter((r) => r.driftType !== "unmatched")
      .reduce((sum, r) => sum + r.similarity, 0) /
    Math.max(results.filter((r) => r.driftType !== "unmatched").length, 1);

  return {
    matchedColumns: results,
    newColumns,
    removedColumns,
    overallDriftScore: 1 - avgSimilarity,
  };
}

The threshold values (0.88, 0.72) are not universal. Calibrate them on a set of known renames and genuine new columns from your own schema history.

---

LLM-Assisted Data Quality Rule Generation

Documentation exists for most data sources: API specs, database ERDs, Confluence pages, README files. LLMs can parse that documentation and generate explicit validation rules, reducing the bootstrap cost for a new source from “write rules manually” to “paste the docs, review the output.”

interface GeneratedRule {
  columnName: string;
  ruleType:
    | "not_null"
    | "range"
    | "enum"
    | "regex"
    | "referential"
    | "custom";
  severity: "warning" | "error";
  parameters: Record<string, unknown>;
  rationale: string;
  confidence: "high" | "medium" | "low";
}

interface RuleGenerationRequest {
  sourceDocumentation: string;
  sampleRows: Record<string, unknown>[];
  existingProfile: DatasetProfile;
}

async function generateRulesFromDocumentation(
  request: RuleGenerationRequest,
  llm: (prompt: string) => Promise<string>
): Promise<GeneratedRule[]> {
  const columnSummaries = request.existingProfile.columns
    .map(
      (col) =>
        `${col.columnName} (${col.dataType}): nullRate=${(col.nullRate * 100).toFixed(1)}%, distinct=${col.distinctCount}`
    )
    .join("\n");

  const prompt = `You are analyzing a data source to generate data quality validation rules.

Source documentation:
${request.sourceDocumentation}

Column statistics from recent data:
${columnSummaries}

Sample rows (first 5):
${JSON.stringify(request.sampleRows.slice(0, 5), null, 2)}

Generate validation rules as a JSON array. Each rule must have:
- columnName: exact column name from the statistics
- ruleType: one of not_null, range, enum, regex, referential, custom
- severity: warning or error
- parameters: rule-specific config (e.g., {"min": 0, "max": 100} for range, {"values": [...]} for enum)
- rationale: one sentence explaining why this rule exists based on the documentation
- confidence: high if the documentation explicitly states the constraint, medium if inferred, low if speculative

Return only the JSON array. No explanation.`;

  const response = await llm(prompt);

  try {
    const rules = JSON.parse(response) as GeneratedRule[];
    // Filter low-confidence rules: require human review before activation
    return rules.filter((r) => r.confidence !== "low");
  } catch {
    throw new Error(`Failed to parse LLM rule output: ${response.slice(0, 200)}`);
  }
}

The output requires human review before rules go active. Treat LLM-generated rules as drafts in a staging state. A rule moves to active after a data engineer confirms it matches the intent of the documentation.

---

Automated Data Contract Enforcement and CI Integration

Data contracts formalize expectations between producers and consumers. They live in version control and gate deployments.

interface DataContract {
  contractVersion: string;
  sourceId: string;
  tableName: string;
  owner: string;
  consumers: string[];
  sla: {
    maxNullRateByColumn: Record<string, number>;
    minRowCount: number;
    maxSchemaChangeScore: number;
  };
  rules: GeneratedRule[];
  activeFrom: string;
}

interface ContractViolation {
  contractVersion: string;
  violationType: "sla_breach" | "rule_violation" | "schema_drift";
  severity: "warning" | "error" | "critical";
  column?: string;
  expected: unknown;
  observed: unknown;
  message: string;
}

async function enforceContract(
  contract: DataContract,
  incomingProfile: DatasetProfile,
  driftResult: SchemaDriftResult,
  signals: AnomalySignal[]
): Promise<ContractViolation[]> {
  const violations: ContractViolation[] = [];

  // Row count SLA
  if (incomingProfile.rowCount < contract.sla.minRowCount) {
    violations.push({
      contractVersion: contract.contractVersion,
      violationType: "sla_breach",
      severity: "critical",
      expected: `>= ${contract.sla.minRowCount} rows`,
      observed: incomingProfile.rowCount,
      message: `Batch row count ${incomingProfile.rowCount} is below minimum ${contract.sla.minRowCount}`,
    });
  }

  // Null rate SLA per column
  for (const col of incomingProfile.columns) {
    const maxNull = contract.sla.maxNullRateByColumn[col.columnName];
    if (maxNull !== undefined && col.nullRate > maxNull) {
      violations.push({
        contractVersion: contract.contractVersion,
        violationType: "sla_breach",
        severity: col.nullRate > maxNull * 2 ? "critical" : "error",
        column: col.columnName,
        expected: `null rate <= ${(maxNull * 100).toFixed(1)}%`,
        observed: `${(col.nullRate * 100).toFixed(1)}%`,
        message: `Column ${col.columnName} null rate ${(col.nullRate * 100).toFixed(1)}% exceeds contract limit ${(maxNull * 100).toFixed(1)}%`,
      });
    }
  }

  // Schema drift SLA
  if (driftResult.overallDriftScore > contract.sla.maxSchemaChangeScore) {
    violations.push({
      contractVersion: contract.contractVersion,
      violationType: "schema_drift",
      severity: "error",
      expected: `drift score <= ${contract.sla.maxSchemaChangeScore}`,
      observed: driftResult.overallDriftScore.toFixed(3),
      message: `Schema drift score ${driftResult.overallDriftScore.toFixed(3)} exceeds contract limit. Removed columns: ${driftResult.removedColumns.join(", ")}`,
    });
  }

  // Anomaly signals that breach contract
  for (const signal of signals) {
    if (signal.severity === "critical" || signal.severity === "high") {
      violations.push({
        contractVersion: contract.contractVersion,
        violationType: "rule_violation",
        severity: signal.severity === "critical" ? "critical" : "error",
        column: signal.columnName,
        expected: signal.baseline,
        observed: signal.observed,
        message: signal.description,
      });
    }
  }

  return violations;
}

For CI integration, contracts are tested against synthetic data representing known good and known bad batches:

// data-contract.test.ts: runs in CI on each contract change
import { describe, it, expect } from "vitest";
import { enforceContract } from "../src/contracts/enforcer";
import { loadContract } from "../src/contracts/loader";
import { goodBatch, nullSpikedBatch, schemaDriftedBatch } from "./fixtures";

describe("orders contract v2.1", () => {
  const contract = loadContract("contracts/orders-v2.1.json");

  it("accepts a valid batch with no violations", async () => {
    const violations = await enforceContract(contract, goodBatch.profile, goodBatch.drift, goodBatch.signals);
    const errors = violations.filter((v) => v.severity === "error" || v.severity === "critical");
    expect(errors).toHaveLength(0);
  });

  it("rejects a batch with null rate spike on order_id", async () => {
    const violations = await enforceContract(contract, nullSpikedBatch.profile, nullSpikedBatch.drift, nullSpikedBatch.signals);
    expect(violations.some((v) => v.column === "order_id" && v.violationType === "sla_breach")).toBe(true);
  });

  it("rejects a batch with schema drift above threshold", async () => {
    const violations = await enforceContract(contract, schemaDriftedBatch.profile, schemaDriftedBatch.drift, schemaDriftedBatch.signals);
    expect(violations.some((v) => v.violationType === "schema_drift")).toBe(true);
  });
});

---

Quarantine Workflow

Data that fails contract enforcement should not flow into production tables. A quarantine workflow holds the batch, notifies the relevant teams, and provides a structured path to either fix-and-reprocess or discard.

type QuarantineAction = "hold" | "discard" | "release_with_warnings";

interface QuarantineRecord {
  batchId: string;
  sourceId: string;
  tableName: string;
  quarantinedAt: string;
  violations: ContractViolation[];
  action: QuarantineAction;
  resolvedAt?: string;
  resolvedBy?: string;
  notes?: string;
}

async function quarantineBatch(
  batchId: string,
  sourceId: string,
  tableName: string,
  violations: ContractViolation[],
  store: QuarantineStore,
  notifier: AlertNotifier
): Promise<QuarantineRecord> {
  const hasCritical = violations.some((v) => v.severity === "critical");
  const action: QuarantineAction = hasCritical ? "hold" : "release_with_warnings";

  const record: QuarantineRecord = {
    batchId,
    sourceId,
    tableName,
    quarantinedAt: new Date().toISOString(),
    violations,
    action,
  };

  await store.save(record);

  if (hasCritical) {
    await notifier.alert({
      channel: `data-quality-${sourceId}`,
      severity: "critical",
      message: `Batch ${batchId} from ${tableName} quarantined. ${violations.filter((v) => v.severity === "critical").length} critical violations.`,
      violations,
    });
  }

  return record;
}

The quarantine store is queryable. When data consumers report missing data, the first lookup should be the quarantine log.

---

Tradeoffs

Decision	Strict	Permissive
Contract violations	Block ingestion on any error	Warn only, let data through
Anomaly threshold	Low z-score (2.5), catch more	High z-score (4+), fewer false positives
Schema drift response	Quarantine on any drift	Alert only unless column removed
LLM rule generation	Human review on every rule	Auto-activate high-confidence rules
Isolation forest scoring	Run on every batch	Sample 10% of batches to reduce cost
Baseline window	90 days (stable)	14 days (adapts faster, noisier)

False positive fatigue is the real risk in strict mode. If quarantine fires five times per week on batches that turn out to be fine, the data team starts ignoring alerts or bypassing quarantine. Calibrate thresholds against three to six months of historical data before activating enforcement in production.

Cost of ML scoring per ingestion is nontrivial when dealing with hundreds of tables at high frequency. Options: run isolation forest on a schedule rather than per-batch, use z-scores for real-time checks and save the ML pass for a daily audit, or gate ML scoring behind a fast pre-filter that only sends anomalous batches (flagged by z-score) to the heavier model.

Embedding calls for schema comparison add latency. Cache embeddings per schema version rather than recomputing on every batch. A schema hash keyed to the embedding avoids redundant API calls.

---

Production Considerations

Baseline staleness: Baselines must update over time or they flag legitimate data evolution as anomalies. Retrain baselines on a weekly schedule, with a human-approval gate for significant baseline changes on high-stakes tables.

Multi-environment contracts: Maintain separate contracts for staging and production. Staging contracts can be permissive enough to allow experimental schema changes without blocking the pipeline. Only production contracts enforce quarantine.

Bootstrapping new sources: A new source has no baseline. Run the first thirty batches in observation-only mode: profile each one, collect the distribution, then compute the baseline after. Enforce contracts only after the baseline is stable.

Cardinality explosions in categorical columns: A column that normally has 10 distinct values suddenly has 10,000. This usually means a free-text field was accidentally placed into what should be an enum column. Z-score on cardinalityRatio catches this, but you need at least a week of baseline data to compute a meaningful standard deviation for cardinality.

Handling legitimate schema changes: Schema migrations should be announced via a contract update PR before the data change lands. The drift detection then sees the expected new schema and does not quarantine. This requires coordination between data producers and the data platform team, which is the organizational part of data contracts that pure tooling cannot solve.

---

Rule-based validation tells you whether a column satisfies a predicate you thought to write. Statistical and embedding-based approaches tell you whether a column’s behavior today is consistent with its history. Both are necessary. The combination is what makes data quality enforceable at scale without a team of engineers manually reviewing every ingestion run.