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

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

1. 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.

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:

1. New sensor temperature is added

2. gps.coordinate format changes from [lat, long] to GeoJSON

3. 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)})

1. Schema Evolution Strategy:

   • Backward Compatibility: New fields have defaults

   • Forward Compatibility: Ignore unknown fields

   • Use Avro’s default and aliases for field renames