How do you validate schemas at ML pipeline boundaries?

Schema validation checks whether data entering or leaving an ML pipeline step matches a declared contract. It catches missing columns, unexpected types, nulls, invalid ranges, unsupported categories, and incompatible model inputs *before* incorrect data reaches training or production inference.

Question 96 covered sharing transformation definitions across training and serving. This lesson focuses on checking the actual data at each handoff, rather than defining how features are computed.

Beginner-friendly example: delivery-time prediction

A delivery model expects the following inputs:

• order_id: required nonempty string.
• distance_km: required numeric value, 0–100.
• traffic_level: required category among low, medium, high.
• restaurant_delay_min: optional numeric, 0–120 when present.
• event_timestamp: required valid timezone-aware timestamp.

A newly ingested batch has 1,000 rows:

• 970 rows satisfy all the rules.
• 12 have missing distance_km.
• 8 have distance_km = -5.
• 10 contain an unexpected traffic_level = extreme.

Assuming these problems occur in distinct rows, there are 30 invalid rows. The valid-row rate is 970 / 1,000 × 100 = 97%, and the invalid-row rate is 3%.

If the pipeline policy requires at least 99% valid rows, the batch fails, even though 970 records look usable. If violations overlap, calculate distinct invalid rows rather than summing failures.

Where should you validate?

1. Ingestion boundary — source to landing area

• Assert expected columns and parseable data types.
• Reject missing keys or malformed timestamps.
• Flag out-of-range values and unexpected category codes.
• Quarantine invalid records with a reason; preserve original input for diagnosis if privacy rules permit.

2. Transformation boundary — raw to features

• Check that calculated features have correct names, types, ranges, null semantics and units.
• For example, distance_km must not silently become meters.
• Validate output row counts and entity-key uniqueness when required by the feature design.

3. Training boundary — features to model fitting

• Check required model features, feature order where applicable, consistent categorical encodings and expected target/label presence.
• Reject data that would silently change the model's input contract.
• Validate label availability and type *only where labels are expected*; labels typically are not supplied by live inference requests.

4. Inference boundary — request to model

• Validate incoming payloads against the deployed model's signature or API schema.
• Return a controlled input error for missing/unsupported fields; don't let a malformed request trigger an unhandled model exception.
• Also verify feature-store lookup outputs after retrieval.

5. Model output boundary

• Validate predicted response structure and type, such as a finite numeric eta_minutes. Check contractual constraints and handle NaN/infinity safely.
• Do not confuse a structurally valid output with an accurate prediction.

What counts as a schema contract?

A practical schema defines:

1. Required columns and keys: Names, optional versus mandatory fields, unexpected column policy and row/entity-key requirements.
2. Types: String, integer, float, timestamp, boolean or supported categorical representation.
3. Nullability: Whether a missing value is permitted and how it is represented.
4. Ranges: For example 0 <= distance_km <= 100, including inclusive-boundary rules.
5. Categories: Known values for traffic_level, plus an explicit unknown-category policy.
6. Semantic constraints: Units, timestamps, uniqueness and cross-field rules.
7. Compatibility: Which producer schema versions a pipeline stage or deployed model accepts.

Schema checks differ from statistical monitoring: A dataset could pass every type/range check while its distribution changes drastically. Drift detection and model-quality monitoring are separate checks.

A simple executable Python validator

PYTHON
import math

ALLOWED_TRAFFIC = {"low", "medium", "high"}

def validate_row(row: dict) -> list[str]:
    errors = []
    if not isinstance(row.get("order_id"), str) or not row["order_id"].strip():
        errors.append("order_id must be a nonempty string")

    d = row.get("distance_km")
    if isinstance(d, bool) or not isinstance(d, (int, float)):
        errors.append("distance_km must be numeric")
    elif not math.isfinite(d) or not (0 <= d <= 100):
        errors.append("distance_km outside [0, 100]")

    if row.get("traffic_level") not in ALLOWED_TRAFFIC:
        errors.append("unexpected traffic_level")

    delay = row.get("restaurant_delay_min")
    if delay is not None:
        if isinstance(delay, bool) or not isinstance(delay, (int, float)):
            errors.append("restaurant_delay_min must be numeric or null")
        elif not math.isfinite(delay) or not (0 <= delay <= 120):
            errors.append("restaurant_delay_min outside [0, 120]")

    return errors

valid = {
    "order_id": "O42",
    "distance_km": 7.5,
    "traffic_level": "medium",
    "restaurant_delay_min": None,
}
invalid = dict(valid, distance_km=-5)
print(validate_row(valid))     # []
print(validate_row(invalid))   # ['distance_km outside [0, 100]']

This intentionally small example covers a subset of the full contract; it does not validate event_timestamp, unexpected keys, provenance, or batch uniqueness. Production pipelines should use an appropriate validation library or structured schema implementation, covered by tests.

Validating a whole dataset with Great Expectations

Great Expectations (GX) supports schema expectations such as required column existence, column data types, ordered column lists, acceptable value sets and numerical ranges.

Conceptually, an ingestion checkpoint might enforce:

order_id: must exist, must not be null
distance_km: numerical, not null, 0–100
traffic_level: low | medium | high
restaurant_delay_min: null allowed; otherwise 0–120
event_timestamp: parseable timestamp

The GX documentation includes ExpectColumnToExist, ExpectColumnValuesToBeOfType, ExpectTableColumnsToMatchSet and range/category expectations. Use versions and APIs supported by your environment; validate the source dataset through a committed, reviewed expectation suite.

TensorFlow Data Validation (TFDV) also detects anomalies by checking dataset statistics against an explicit schema. Its training/serving schema environments let a label column be required for training but absent in serving without raising a false error.

Backward-compatibility decisions

A producer adding an optional field can be safe if consumers tolerate unknown columns. Removing a required field, changing distance_km from float to string, or changing units from kilometers to meters can break consumers.

• Compatible addition: New nullable column that old consumers ignore.
• Potentially breaking change: Renaming an input column, altering its type/units, or making an optional field required.
• Category extension: A new traffic_level category may be structurally representable yet unsupported by the model's encoder.
• Model-specific contract: The same source schema can feed different models with different required subsets of columns.

Always test compatibility against actual downstream versions; do not declare a change safe solely because it is additive. Question 99 covers schema evolution in greater detail.

What happens when validation fails?

• Fail closed for critical fields: Stop training or reject the inference request when required inputs cannot be trusted.
• Quarantine records: Isolate invalid data with concise reason codes and auditable batch IDs.
• Apply an approved fallback: Only for explicitly optional fields and models tested with those fallback values.
• Notify the owner: Alert the upstream producer or pipeline owner with counts, failure categories and sample references that do not expose sensitive data.
• Retry appropriately: Retry transient read/service failures, not deterministic schema incompatibilities.
• Prevent partial publication: Do not mark a dataset or artifact validated if critical checks fail.

Do not automatically coerce everything: blindly converting malformed strings or out-of-range values may hide upstream defects.

Production test checklist

• Run golden valid and intentionally invalid datasets through every relevant boundary.
• Test missing columns, extra columns, unexpected categories, nullable fields, NaN/infinity, units and timestamp parsing.
• Verify label-required training schemas are distinguished from label-free serving schemas.
• Assert reject, quarantine, alert and fallback behaviors.
• Track validation-failure rate, affected entities, schema version, release ID and top failure types.
• Recheck the model's actual serving signature before deployment.
• Version the schema with the pipeline code and pin the schema accepted by each model.

In short: Put explicit, versioned schema checks at ingestion, feature transformation, training, inference and output boundaries. Reject unsafe incompatibilities early, record clear failures, and test compatibility for every pipeline consumer.