What’s the best way to handle schema evolution in a data pipeline?

Question

Tuhin Paul · Answer

Part -2 Real-Time Transformation : # Use SQL-based stream processing for schema upgrades CREATE STREAM upgraded_telemetry AS SELECT timestamp STRUCT(type := 'Point' coordinates := ARRAY[gps[0] gps[1]]) AS gps COALESCE(temperature 25.0) AS temperature lidar FROM raw_telemetry AI Model Versioning : Serve multiple model versions via TFX Pipelines Use schema version → model version mapping This architecture validates zero downtime schema changes while maintaining AI processing capabilities for critical drone operations.

Tuhin Paul · Answer

Part - 1 I will try to share a detailed solution for schema evolution in my recent project on anomaly detection pipeline using schema versioning and backward compatibility patterns. This example will be bit advance in nature. A critical system processes live drone sensor data (GPS accelerometer lidar) to detect anomalies. The schema evolves when: New sensor temperature is added gps.coordinate format changes from [lat long] to GeoJSON Field lidar.resolution is deprecated Schema Registry (Avro) # schemas/telemetry.avsc (v3) { type: record name: DroneTelemetry fields: [ {name: timestamp type: long} {name: gps type: {type: record name: GeoJSON fields: [{name: type type: string} {name: coordinates type: {type: array items: double}}]}} {name: temperature type: float default: 25.0} {name: lidar type: {type: array items: float}} ] } Schema Transformer (Python with FastAPI) from dataclasses import dataclass from fastapi import APIRouter from aiokafka import AIOKafkaConsumer router = APIRouter() @dataclass class TelemetryV1: timestamp: int gps: list[float] # [lat long] lidar: list[float] @dataclass class TelemetryV3: timestamp: int gps: dict # {type: Point coordinates: [lat long]} temperature: float = 25.0 lidar: list[float] async def v1_to_v3(record: TelemetryV1) -> TelemetryV3: return TelemetryV3( timestamp=record.timestamp gps={type: Point coordinates: record.gps} lidar=record.lidar ) @router.post(/process) async def handle_telemetry(data: dict): version = data.get(schema_version 1) match version: case 1: transformed = v1_to_v3(TelemetryV1(**data)) case 2: # Handle v2?v3 transformation pass case 3: transformed = TelemetryV3(**data) # Send to TensorFlow Serving for anomaly detection await detect_anomalies(transformed) Kafka Consumer with Schema Awareness consumer = AIOKafkaConsumer( drone-telemetry bootstrap_servers=localhost:9092 value_deserializer=lambda v: json.loads(v).get(payload) auto_offset_reset=earliest ) async def consume(): async for msg in consumer: schema_version = msg.headers.get(schema_version b1).decode() await router.handle_telemetry({**msg.value schema_version: int(schema_version)}) Schema Evolution Strategy : Backward Compatibility : New fields have defaults Forward Compatibility : Ignore unknown fields Use Avro’s default and aliases for field renames