Automated Data Quality & Anomaly Detection with AI

+---------------------------------------------------------------------------------+
|                       AI-POWERED DATA QUALITY LIFECYCLE                         |
+---------------------------------------------------------------------------------+
|  1. Statistical Profiling -> Mean, Variance, Null %                             |
|  2. Semantic Drift Scan   -> Detect unexpected category values or format drifts |
|  3. Root Cause Analysis   -> LLM analyzes failed pipeline logs & suggests patch |
+---------------------------------------------------------------------------------+

Why Rule-Based Testing (Great Expectations, dbt tests) Falls Short

Traditional data quality tools check hard rules:

• column IS NOT NULL
• column > 0
• column IN ('standard', 'premium')

However, modern data breaks in semantic ways:

• A column intended for currency names starts receiving "USD", "$US", "United States Dollars", and "US Dollar" simultaneously.
• A user comments column suddenly contains spam URLs or prompt injection payloads.
• Distributional drift occurs without triggering a null or range violation.

LLM-Powered Root-Cause Analysis (RCA)

When a critical Airflow DAG or dbt test suite fails at 2:00 AM, an automated AI agent can ingest the stack trace, the previous run logs, and the git diff to produce an immediate triage summary:

from pydantic import BaseModel
from openai import OpenAI

class PipelineFailureRCA(BaseModel):
    summary: str
    probable_root_cause: str
    impacted_downstream_tables: list[str]
    suggested_fix: str
    urgency_level: str

def generate_pipeline_rca(stack_trace: str, dbt_manifest_snippet: str) -> PipelineFailureRCA:
    client = OpenAI()
    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        response_format=PipelineFailureRCA,
        messages=[
            {"role": "system", "content": "You are a Principal Data Engineer on-call. Analyze the following pipeline failure log and provide deterministic root cause analysis."},
            {"role": "user", "content": f"LOG TRACE:
{stack_trace}

CONTEXT:
{dbt_manifest_snippet}"}
        ]
    )
    return response.choices[0].message.parsed

Semantic Anomaly Detection on Column Values

# Semantic check for anomalous address fields
sample_records = [
    "123 Market St, San Francisco, CA 94105",
    "456 Elm Ave, Austin, TX 78701",
    "DROP TABLE users; --",   # Suspicious / Malicious
    "N/A - Customer refused to provide",  # Garbage value passing NOT NULL
    "Flat 4B, Baker St, London, UK"
]

prompt = f"""
Inspect the following text entries destined for an 'address' column.
Identify any entry that is garbage, malicious, an attempted SQL injection, or not a physical address:

{chr(10).join(f"{i+1}. {r}" for i, r in enumerate(sample_records))}
"""