Handling Schema Evolution and Breaking Changes

Schema changes (column additions, deletions, type changes) are not being tracked or communicated effectively, leading to downstream pipeline failures. Organizations need systematic ways to detect, communicate, and manage schema evolution.

You Might Be Asking:

• How does DataHub track schema changes?
• Can I get alerts for breaking schema changes?
• How do I see schema version history?

1. Enable schema change detection in ingestion:

Example for Snowflake. This can be adapted for other data sources/connectors.

source:
  type: snowflake
  config:
    # Required for schema tracking
    include_tables: true
    include_views: true

    # Stateful ingestion tracks changes
    stateful_ingestion:
      enabled: true

    # Emit schema metadata
    include_technical_schema: true

2. Create schema change assertion:

mutation createSchemaAssertion {
  upsertDatasetFieldAssertions(
    input: {
      entityUrn: "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table,PROD)"
      assertions: [
        {
          type: FIELD_ASSERTION
          fieldPath: "user_email"
          assertionInfo: {
            type: "FIELD_VALUES"
            description: "User email must not be null"
          }
        }
      ]
    }
  )
}

3. Set up schema change notifications:

# DataHub Actions configuration
action:
  type: "schema_change_alert"
  config:
    # Monitor for schema changes
    filters:
      - type: "schema_change"
        change_types: ["FIELD_REMOVED", "FIELD_TYPE_CHANGED"]

    # Notification channel
    slack_webhook: "${SLACK_WEBHOOK}"

    # Message template
    message: |
      🚨 Breaking Schema Change Detected
      Dataset: {{dataset.name}}
      Change: {{change.type}}
      Field: {{change.fieldPath}}
      Old Type: {{change.oldType}}
      New Type: {{change.newType}}

4. Query schema history via API:

query {
  dataset(urn: "YOUR_URN") {
    schemaMetadata(version: 0) {
      # Latest schema
      fields {
        fieldPath
        nativeDataType
        description
      }
    }
    # Schema history
    aspects(input: {aspectName: "schemaMetadata", maxCount: 10}) {
      aspects {
        version
        created {
          time
        }
      }
    }
  }
}

5. Document breaking changes:

Example for Snowflake. This can be adapted for other data sources/connectors.

from datahub.emitter.mce_builder import make_dataset_urn
from datahub.metadata.schema_classes import DatasetPropertiesClass

# Add documentation about schema change
properties = DatasetPropertiesClass(
    description="Updated schema - removed deprecated user_age column",
    customProperties={
        "schema_version": "2.0",
        "breaking_change_date": "2025-01-15",
        "migration_guide": "https://docs.company.com/migration-v2"
    }
)

emitter.emit_mcp(
    entity_urn=make_dataset_urn("snowflake", "db.schema.table"),
    aspect_name="datasetProperties",
    aspect=properties
)

6. Implement schema compatibility policies:

Best Practices:
1. Version your schemas explicitly
2. Use schema registries (Avro, Protobuf) where possible
3. Document all breaking changes
4. Provide migration paths
5. Alert downstream consumers before changes
6. Use DataHub assertions to validate schema contracts

Schema evolution tracking requires stateful ingestion to detect changes between runs. Consider implementing schema approval workflows for production datasets. Use tags like "schema-v2" to mark major schema versions.

• Schema Metadata