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", b"1").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

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

Insert Link

Embed YouTube Video

Table Options

Insert Image

Answers (2)