Real-time data validation<\/strong> is the practice of checking incoming data as it arrives<\/em> (streaming or micro-batch) to ensure it is:<\/p>\n Instead of discovering problems after training degrades, real-time validation acts as a gatekeeper<\/strong>:<\/p>\n Source \u2192 Ingestion \u2192 Validation Gate \u2192 (Accept | Quarantine | Reject) \u2192 Storage\/Feature Store \u2192 Training\/Inference<\/strong><\/p>\n Validate structure before<\/strong> any ML preprocessing.<\/p>\n Checks:<\/p>\n Action:<\/p>\n Validate \u201cphysics\/business logic\u201d:<\/p>\n Examples:<\/p>\n Action:<\/p>\n Validate distribution changes and outliers:<\/p>\n Streaming metrics:<\/p>\n Common drift measures:<\/p>\n Action:<\/p>\n If your pipeline includes real-time labels (human feedback, weak supervision, delayed outcomes):<\/p>\n Checks:<\/p>\n Action:<\/p>\n Every event gets a decision:<\/p>\n This prevents one bad stream from contaminating training data.<\/p>\n Treat validation rules as code:<\/p>\n This supports audits and rollback.<\/p>\n For new sources or changes:<\/p>\n This reduces false positives and breaks fewer production flows.<\/p>\n Use a 3-step method:<\/p>\n Tip: false positives are costly. Quarantine is usually better than reject for \u201cmaybe-bad\u201d events.<\/p>\n<\/blockquote>\n from dataclasses import dataclassfrom typing import Dict, Any, Tuple, Listimport mathimport re@dataclassclass ValidationResult: decision: str # “ACCEPT” | “QUARANTINE” | “REJECT” reasons: List[str] rule_version: str = “v1.0″def validate_event(event: Dict[str, Any]) -> ValidationResult: reasons = [] # —- Schema checks (hard) —- required = [“user_id”, “amount”, “currency”, “timestamp”] for f in required: if f not in event: return ValidationResult(“REJECT”, [f”Missing required field: {f}”]) if not isinstance(event[“user_id”], str) or len(event[“user_id”]) < 3: return ValidationResult(“REJECT”, [“Invalid user_id type\/length”]) if not isinstance(event[“amount”], (int, float)): return ValidationResult(“REJECT”, [“amount must be numeric”]) # —- Constraint checks (hard\/soft) —- amt = float(event[“amount”]) if math.isnan(amt) or math.isinf(amt): return ValidationResult(“REJECT”, [“amount is NaN\/Inf”]) if amt < 0: return ValidationResult(“REJECT”, [“amount < 0 is invalid”]) if amt > 1_000_000: reasons.append(“amount unusually high (soft)”) cur = event[“currency”] if cur not in {“USD”, “EUR”, “INR”, “GBP”}: return ValidationResult(“REJECT”, [f”Unexpected currency: {cur}”]) # Timestamp format check (example) if not re.match(r”^d{4}-d{2}-d{2}T”, str(event[“timestamp”])): reasons.append(“timestamp format unexpected (soft)”) # —- Drift\/anomaly hooks (soft) —- # Here you’d call monitoring functions (PSI, missingness spikes, etc.) # If anomaly score > threshold -> quarantine. if reasons: return ValidationResult(“QUARANTINE”, reasons) return ValidationResult(“ACCEPT”, [])<\/p>\n PSI compares current distribution to baseline bins. In real-time you compute PSI per window (e.g., last 10k events).<\/p>\n import numpy as npdef psi(expected_counts, actual_counts, eps=1e-6): expected = np.array(expected_counts, dtype=float) + eps actual = np.array(actual_counts, dtype=float) + eps expected = expected \/ expected.sum() actual = actual \/ actual.sum() return float(np.sum((actual – expected) * np.log(actual \/ expected)))<\/p>\n Operational rule example:<\/strong><\/p>\n 0.25 \u2192 high drift (quarantine + alert)<\/p>\n<\/blockquote>\n<\/li>\n<\/ul>\n (Exact thresholds depend on domain; tune them empirically.)<\/p>\n For each event (or batch window), log:<\/p>\n \n \n This enables:<\/p>\n\n
2) What threats real-time validation helps stop<\/h3>\n
\n
3) The validation stack (recommended layered approach)<\/h2>\n
Layer A \u2014 Schema validation (hard gate)<\/h3>\n
\n
\n
Layer B \u2014 Constraint validation (hard + soft gates)<\/h3>\n
\n
\n
Layer C \u2014 Statistical validation (drift + anomaly gates)<\/h3>\n
\n
\n
\n
Layer D \u2014 Label integrity checks (when labels arrive)<\/h3>\n
\n
\n
4) Real-time validation patterns (how to implement operationally)<\/h2>\n
Pattern 1 \u2014 \u201cAccept \/ Quarantine \/ Reject\u201d<\/h3>\n
\n
Pattern 2 \u2014 Validation as code (versioned)<\/h3>\n
\n
Pattern 3 \u2014 \u201cShadow validation\u201d<\/h3>\n
\n
5) Concrete validation rules by data type<\/h2>\n
A) Tabular \/ transactions<\/h3>\n
\n
\n
B) Text (NLP)<\/h3>\n
\n
\n
C) Images<\/h3>\n
\n
6) Thresholding strategy (how to set limits safely)<\/h2>\n
\n
\n
\n
7) Implementation examples (code patterns)<\/h2>\n
7.1 Simple Python \u201cgate\u201d function (schema + constraints + drift hooks)<\/h3>\n
7.2 Streaming drift check idea (PSI skeleton)<\/h3>\n
\n
\n
8) What to log for audits and investigations<\/h2>\n
\n
\n