AI Data Architecture, Part 3: Data Quality Automation and Governance

from dataclasses import dataclass
from typing import Callable

@dataclass
class QualityRule:
    name: str
    check: Callable
    severity: str  # "error" | "warning"
    description: str

class DataQualityFramework:
 """AI """

    def __init__(self):
        self.rules: list[QualityRule] = []

    def add_rule(self, rule: QualityRule):
        self.rules.append(rule)

    def validate(self, df) -> dict:
        results = {"passed": [], "failed_errors": [], "failed_warnings": []}

        for rule in self.rules:
            try:
                passed = rule.check(df)
                if passed:
                    results["passed"].append(rule.name)
                elif rule.severity == "error":
                    results["failed_errors"].append({
                        "rule": rule.name,
                        "description": rule.description,
                    })
                else:
                    results["failed_warnings"].append(rule.name)
            except Exception as e:
                results["failed_errors"].append({"rule": rule.name, "error": str(e)})

        results["pipeline_blocked"] = len(results["failed_errors"]) > 0
        return results

# example
qf = DataQualityFramework()
qf.add_rule(QualityRule(
    name="null_check_embedding",
    check=lambda df: df["embedding"].notna().all(),
    severity="error",
 description=" NULL ",
))
qf.add_rule(QualityRule(
    name="freshness_check",
    check=lambda df: (pd.Timestamp.now() - df["updated_at"].max()).hours < 24,
    severity="warning",
 description="24when ",
))

class LineageTracker:
    def record_dataset_creation(self, dataset_id: str, sources: list[str], transform: str):
        self.db.insert("lineage_nodes", {
            "node_id": dataset_id,
            "node_type": "dataset",
            "sources": sources,
            "transform_description": transform,
            "created_at": datetime.utcnow().isoformat(),
            "created_by": get_current_user(),
        })

    def record_model_training(self, model_id: str, dataset_ids: list[str], params: dict):
        self.db.insert("lineage_nodes", {
            "node_id": model_id,
            "node_type": "model",
            "sources": dataset_ids,
            "training_params": params,
            "created_at": datetime.utcnow().isoformat(),
        })

    def get_upstream(self, node_id: str, depth: int = 5) -> dict:
 """ / upstream """
        return self.db.recursive_query(
            start=node_id,
            relation="sources",
            max_depth=depth,
        )

import numpy as np
from scipy import stats

class StreamingFeatureMonitor:
    def __init__(self, baseline_stats: dict, window_size: int = 1000):
        self.baseline = baseline_stats
        self.window = []
        self.window_size = window_size
        self.alerts = []

    def ingest(self, features: dict):
        self.window.append(features)
        if len(self.window) >= self.window_size:
            self._check_drift()
 self.window = self.window[self.window_size // 2:] #

    def _check_drift(self):
        for feat_name, baseline in self.baseline.items():
            current_vals = [w[feat_name] for w in self.window if feat_name in w]
            if not current_vals:
                continue

 # Kolmogorov-Smirnov
            ks_stat, p_value = stats.ks_2samp(baseline["samples"], current_vals)

            if p_value < 0.05:
                self.alerts.append({
                    "feature": feat_name,
                    "ks_statistic": ks_stat,
                    "p_value": p_value,
                    "baseline_mean": baseline["mean"],
                    "current_mean": np.mean(current_vals),
                    "severity": "critical" if ks_stat > 0.3 else "warning",
                })

Centralized (traditional)          Data mesh (recommended)
─────────────────────         ─────────────────────
  Data team → single hub         Marketing team → marketing data product
  (bottleneck, slow response)    Search team    → search data product
                                 Orders team    → orders data product
                                      ↓
                               Global governance layer
                               (standard schema + quality SLA)

class DataProduct:
 """ when """
    def __init__(self, domain: str, owner_team: str):
        self.domain = domain
        self.owner = owner_team
        self.quality_sla = {"freshness_hours": 1, "completeness": 0.99}

    def publish(self, data, schema_version: str):
        self.validator.check(data, schema_version)
        self.catalog.register(
            product_id=f"{self.domain}/v{schema_version}",
            owner=self.owner,
            quality_metrics=self.measure_quality(data),
        )
        self.storage.write(data)