How do you handle schema evolution without breaking a training pipeline?

Schema evolution means changing the structure or interpretation of upstream data while keeping existing training jobs and model-serving consumers working. The safest approach is to make compatible changes first, version incompatible changes explicitly, test old and new consumers, then roll out gradually with a rollback path.

Question 98 covered the producer–consumer data contract itself. This lesson focuses on how to migrate an existing schema without interrupting training.

Beginner-friendly example

A food-delivery training pipeline expects:

• order_id: string.
• distance_km: float, measured in kilometers.
• delivery_duration_min: float, measured in minutes.

The Orders team wants to introduce a useful feature, weather_code, and later rename distance_km to distance_meters.

Change A — add an optional field: Adding nullable weather_code can be backward-compatible for old consumers if they ignore extra fields. New consumers should define how to handle historical rows where this feature is missing.

Change B — rename and change units: Replacing distance_km with distance_meters is potentially breaking. A legacy training step expecting distance_km will fail or, worse, consume meters as kilometers if the name or semantics are confused.

For a 7.5 km delivery:

• Old representation: distance_km = 7.5.
• New representation: distance_meters = 7,500.
• Without conversion, an old model expecting kilometers could see an input 1,000 times larger.
• Correct adapter: distance_km = distance_meters / 1000 = 7.5.

Notice the risk: a syntactically valid float is not proof that units and meanings are compatible.

Backward versus forward compatibility

For schemas used by a producer and a consumer:

• Backward compatibility: A reader using the *new schema* can consume data written with the *old schema*.
• Forward compatibility: A reader using the *old schema* can consume data written with the *new schema*.
• Full compatibility: Both directions are supported.
• Transitive compatibility: The selected rule is checked against all supported previous schema versions, not just the most recent one.

These definitions match Confluent Schema Registry documentation. Actual allowable changes depend on serialization format (Avro, Protobuf, JSON Schema), defaults, reader implementation and pipeline code. For example, adding a nullable/defaulted field is often safe in Avro; adding a required field with no valid default can break readers of older records.

For ML pipelines, compatibility also involves business meaning, model expectations and units—which basic serialization schema compatibility checks may miss.

Migration pattern: expand → migrate → contract

1. Expand: Introduce a new optional field, or publish both old and new fields during a transition. Keep existing readers functional.
2. Adapt: Add a tested conversion in the ingestion or feature transformation boundary. Record which input schema version was used.
3. Migrate: Upgrade training pipelines and deployed models to consume the new canonical representation.
4. Verify: Compare values and retraining results on representative old and new data.
5. Contract: Remove the old field only after every supported consumer has migrated and old dataset replays have a valid compatibility path.

For example, publish both distance_km and distance_meters temporarily. Validate that abs(distance_meters / 1000 - distance_km) is within an approved numerical tolerance, and alert on mismatches. Avoid permanent dual-write complexity: define a deprecation deadline and keep a plan for older stored data.

Simple Python adapter

PYTHON
import math

def normalize_distance(record: dict) -> float:
    if "distance_km" in record and record["distance_km"] is not None:
        km = float(record["distance_km"])
    elif "distance_meters" in record and record["distance_meters"] is not None:
        km = float(record["distance_meters"]) / 1000.0
    else:
        raise ValueError("Missing distance")

    if not math.isfinite(km) or km < 0 or km > 100:
        raise ValueError("Invalid distance in km")
    return km

assert normalize_distance({"distance_km": 7.5}) == 7.5
assert normalize_distance({"distance_meters": 7500}) == 7.5

This illustrates conversion, not a complete production adapter. If both fields are present, real code should check that they agree instead of automatically trusting one. Also reject booleans and invalid type coercions according to your data contract.

Defaults: useful, but not magic

Suppose the new training pipeline expects weather_code, but six months of historical records have no weather value.

• A nullable field and explicit unknown category may be appropriate if the model is trained to handle them.
• Filling every missing value with sunny would invent weather observations and may distort training.
• An Avro field default helps schema resolution; it does not prove that substituting the default is statistically safe for the ML model.
• For retraining with historical data, a backfill from a trustworthy historical weather source may be better, if available.

Document when defaults are used and test model performance on those cases.

Compatibility tests before deployment

Producer tests

• New records satisfy their declared schema and required semantic constraints.
• Old fields remain available during the compatibility period.
• Conversion between units is accurate across boundary and typical cases.
• No unsupported null/category values are introduced.

Consumer tests

• The *current* production training job still reads new producer output.
• The *new* training job reads both old stored records and new records.
• Model feature order, dtype and units remain what the trained model expects.
• An expected failure blocks promotion and produces a clear diagnostic.

Historical replay

• Load representative snapshots from previous schema versions.
• Check the normalized features for identical business meaning.
• Train or dry-run a small pipeline on old and new snapshots; compare feature statistics and metrics.
• Preserve version IDs and transformation code revisions to explain any difference.

A staged rollout example

Suppose there are 10 downstream training pipelines depending on distance_km.

1. Begin dual-field publication without removing distance_km.
2. Upgrade 2 pipelines to the v2 adapter and run them with shadow validation.
3. If results match, upgrade the remaining 8 pipelines in controlled batches.
4. Confirm that 10/10 registered consumers use compatible logic and historical replays work.
5. Only after a documented retirement window, stop emitting the old field.

If the first two pipelines show unexpected feature changes, revert the consumer adapter while keeping the old producer field available. A successful rollout is not measured only by whether code deploys—it must preserve the model input contract.

Rollback planning

• Before rollout: Save the old schema definition, producer deployment, adapter version and approved dataset snapshots.
• During rollout: Keep a dual-field or backward-readable path for older consumers.
• On failure: Pause the cutover, restore compatible producer or consumer configuration, and rerun affected pipeline jobs using trusted snapshots.
• After rollback: Verify that ingestion, transformation, training and model predictions receive the intended values.
• Long-term: Do not drop old schema support until data retention and replay requirements no longer depend on it.

A schema registry may reject incompatible wire-format changes, but it cannot automatically roll back producer semantics or retrain a model.

Common mistakes

• Renaming a field without updating historical datasets and downstream consumers.
• Changing kilometers to meters while keeping the same feature name.
• Treating nullable/defaulted fields as automatically harmless for model quality.
• Deploying a new producer before checking supported old readers.
• Testing only the newest schema instead of replaying older snapshots.
• Keeping two conflicting definitions indefinitely without ownership.
• Deleting deprecated columns before confirming all consumers have moved.
• Assuming a BACKWARD setting checks every historical schema version; Confluent's default BACKWARD is non-transitive unless explicitly configured otherwise.

In short: Evolve schemas with versioned changes, compatibility checks, explicit unit/semantic adapters, gradual consumer upgrades, historical replay tests and a reversible rollout. Do not confuse structural compatibility with ML feature correctness.