Schema Aware Guardrails for Real Time Feature Tables

The problem I ran into: “valid JSON” that still broke training

I built a real-time feature pipeline for an AI model and quickly learned a painful lesson: a payload can be “valid JSON” and still be invalid for your model.

Here’s what happened in my setup:

• Events arrive continuously (think clickstream, sensor readings, etc.).
• A streaming job converts those events into a feature table used by the model.
• The model (or training job) expects very specific fields:
  • certain columns must exist
  • certain columns must be numeric
  • timestamps must be parseable
  • “required” fields can’t be null
• After a deploy, a producer started sending "amount": "12.34" (string) instead of 12.34 (number).
  The JSON parse succeeded, but downstream feature computation produced nulls and the model training silently degraded.

That’s an AI Data Governance failure: data violated the agreed contract, but there was no automated way to detect it early.

A very specific guardrail I implemented: a schema contract on feature rows

To stop this class of failure, I implemented schema-aware guardrails directly in the streaming-to-feature step.

The idea is simple and strict:

1. Define a contract for every feature row (types + required fields).
2. Validate each incoming event against the contract.
3. If validation fails:
   • route the row to a quarantine stream/table
   • record why it failed
   • never let it reach the “training-ready” feature table

I used Python + Pandera (data validation library for pandas) to make this concrete and executable.

---

Example setup: event → feature row

Assume incoming events look like this:

{
  "user_id": 42,
  "event_ts": "2026-11-13T10:11:12Z",
  "amount": "12.34",
  "currency": "USD"
}

The model’s feature contract expects:

• user_id: integer
• event_ts: timezone-aware datetime
• amount: float (not string)
• currency: non-empty string

Step 1: Define the feature-row schema contract

import pandera as pa
from pandera import Field
from pandera.typing import Series
import pandas as pd

feature_schema = pa.DataFrameSchema(
    {
        "user_id": Series[int] .astype(int),
        "event_ts": Series[pd.Timestamp],
        "amount": Series[float],
        "currency": Series[str],
    },
    checks=[
        # currency must not be empty/whitespace
        pa.Check(lambda df: df["currency"].str.strip().ne(""), element_wise=False),
    ],
    strict=True,      # no extra columns
    coerce=False,     # do not auto-convert strings to numbers (important!)
)

Why these settings matter:

• strict=True prevents producers from adding unexpected columns that could mask issues.
• coerce=False is the key: I want failures when types don’t match. Auto-coercion would hide the exact problem we’re trying to detect.

---

Step 2: Build a validator that returns “clean features” + “quarantine reasons”

Here’s the core workflow:

• Parse raw events into a DataFrame
• Validate against the schema
• Split into:
  • features_clean (valid rows)
  • quarantine (invalid rows + error message)

def validate_feature_rows(events: list[dict]):
    df = pd.DataFrame(events)

    # Normalize obvious structural issues for better error messages
    if df.empty:
        return df, pd.DataFrame(columns=["event", "error"])

    try:
        validated = feature_schema.validate(df)
        quarantine = pd.DataFrame(columns=["event", "error"])
        return validated, quarantine
    except pa.errors.SchemaErrors as e:
        # Pandera reports failure details; we’ll quarantine all rows that failed.
        # For row-level quarantine in a streaming app, you'd run per-row validation.
        error_df = e.failure_cases.copy()
        quarantine = pd.DataFrame(
            {
                "event": df.to_dict(orient="records"),
                "error": str(e)
            }
        )
        # Keep only rows that passed (best-effort via boolean mask if available)
        # In Pandera SchemaErrors, failure_cases doesn't always map cleanly to row indices,
        # so I take a simple approach here.
        return pd.DataFrame(columns=df.columns), quarantine

This example quarantines at the batch level for clarity. In production, I validated per-row so quarantine is precise.

---

Step 3: Run it on “good” and “bad” events

good_events = [
    {"user_id": 42, "event_ts": "2026-11-13T10:11:12Z", "amount": 12.34, "currency": "USD"},
]

bad_events = [
    # amount is a string, not a float
    {"user_id": 42, "event_ts": "2026-11-13T10:11:12Z", "amount": "12.34", "currency": "USD"},
    # currency is empty
    {"user_id": 7, "event_ts": "2026-11-13T10:11:12Z", "amount": 5.0, "currency": "   "},
]

good_features, good_quarantine = validate_feature_rows(good_events)
bad_features, bad_quarantine = validate_feature_rows(bad_events)

print("GOOD FEATURES")
print(good_features)
print("\nGOOD QUARANTINE")
print(good_quarantine)

print("\nBAD FEATURES")
print(bad_features)
print("\nBAD QUARANTINE")
print(bad_quarantine[["error"]].head())

What happens when I ran this

• The good_events passed and produced a validated feature DataFrame.
• The bad_events failed schema validation and produced an empty bad_features DataFrame.
• bad_quarantine contained error messages including the schema mismatch (like amount being the wrong type, and currency being empty).

This is exactly the governance behavior I wanted: training-ready tables never receive contract-violating rows.

---

Step 4: Make it row-level (streaming-friendly)

Batch-level quarantine is sometimes too coarse for real-time governance. For streaming, I validated each event independently so I could quarantine specific offenders.

def validate_event(event: dict):
    df = pd.DataFrame([event])

    # Convert timestamps explicitly before schema validation.
    # This keeps type errors distinct from timestamp parsing errors.
    if "event_ts" in df.columns:
        df["event_ts"] = pd.to_datetime(df["event_ts"], utc=True, errors="raise")

    try:
        validated = feature_schema.validate(df)
        return validated.iloc[0].to_dict(), None
    except Exception as e:
        return None, str(e)

def validate_stream(events: list[dict]):
    clean_rows = []
    quarantine_rows = []

    for event in events:
        clean, err = validate_event(event)
        if clean is not None:
            clean_rows.append(clean)
        else:
            quarantine_rows.append({"event": event, "error": err})

    return pd.DataFrame(clean_rows), pd.DataFrame(quarantine_rows)

Why parse timestamps explicitly:

• Timestamp parsing failures are a governance issue too (producers sending malformed timestamps).
• Separating parsing errors from type mismatches makes the quarantine easier to triage.

---

Step 5: Integrate governance into your feature table write

In a real pipeline, “write” usually means inserting into a storage table.

Here’s a simplified version showing the governance routing:

def write_feature_tables(features: pd.DataFrame, quarantine: pd.DataFrame):
    # In production, replace these with writes to your warehouse / lake / streaming sinks.
    print(f"Writing {len(features)} rows to training_feature_table")
    print(f"Writing {len(quarantine)} rows to quarantine_feature_table")
    print("\nSample quarantine rows:")
    if len(quarantine) > 0:
        print(quarantine.head(3)[["error", "event"]])

And then:

all_events = good_events + bad_events
features, quarantine = validate_stream(all_events)
write_feature_tables(features, quarantine)

This pattern is the governance backbone:

• training table receives only contract-compliant feature rows
• quarantine table captures the “why”
• observability becomes possible because you can alert on quarantine rates

---

What this governance guardrail buys you (practically)

After implementing this schema-aware contract validation, I could answer governance questions like:

• Did the producer change a field type?
• Did a required field start arriving empty?
• Are timestamp formats drifting over time?
• Are we seeing a spike in quarantine events after a deploy?

Most importantly: model training stopped degrading silently, because contract violations were never allowed into the training-ready feature table.

---

Conclusion

I built schema-aware guardrails that validate real-time feature rows against a strict contract (types, required fields, and business constraints), then route violations into a quarantine table with actionable error details. The key lesson was that “valid JSON” isn’t the same as “valid AI data,” and enforcing the feature-row contract at ingestion time is a concrete, effective AI Data Governance pattern for preventing silent training failures.