Schema Evolution Toolkit

Schema Evolution Toolkit

Detect, validate, migrate, and analyze schema changes across Delta Lake tables — safely and automatically.

By Datanest Digital | Version 2.0.0 | $39

What You Get

A complete toolkit for managing schema evolution in Databricks / Delta Lake pipelines:

• Schema Detector — compare live table schemas against expected definitions, detect drift
• Schema Diff — detailed field-level diff between any two schema versions (no Spark required)
• Schema Migrator — apply safe migrations (add columns, widen types, rename) with rollback
• Migration Generator — auto-generate SQL forward, rollback, and Python migration scripts
• Impact Analyzer — check which pipelines, views, and dashboards break before deploying
• Compatibility Checker — verify backward/forward compatibility before deploying changes
• Schema Registry — Delta-table-backed registry for versioned schema definitions
• Schema Validator — validate DataFrames against registered schemas at runtime
• Ready-to-use Notebooks — detect drift, evolve schemas, and generate migrations interactively
• Schema Versions — example customer (v1/v2/v3) and order (v1/v2) JSON schema files
• Evolution Strategy Guide — comprehensive guide to schema versioning, migration, and impact analysis

File Tree

schema-evolution-toolkit/
├── README.md
├── manifest.json
├── LICENSE
├── src/
│  ├── schema_detector.py    # Detect schema drift between expected and actual
│  ├── schema_diff.py      # Field-level diff with severity classification
│  ├── schema_migrator.py    # Apply safe migrations with rollback support
│  ├── migration_generator.py  # Generate SQL/Python migration scripts from diffs
│  ├── impact_analyzer.py    # Analyze downstream impact of schema changes
│  ├── compatibility_checker.py # Backward/forward compatibility validation
│  ├── schema_registry.py    # Delta-table-backed schema version registry
│  └── schema_validator.py   # Runtime DataFrame schema validation
├── configs/
│  ├── schema_policy.yaml    # Evolution rules and policies per layer
│  └── schemas/
│    ├── v1_customer.json   # Customer schema v1 (4 fields)
│    ├── v2_customer.json   # Customer schema v2 (rename, add nested struct)
│    ├── v3_customer.json   # Customer schema v3 (arrays, nested structs, new fields)
│    ├── v1_order.json    # Order schema v1 (7 fields)
│    └── v2_order.json    # Order schema v2 (type widening, shipping, discounts)
├── notebooks/
│  ├── detect_drift.py     # Interactive drift detection notebook
│  ├── evolve_schema.py     # Interactive schema migration notebook
│  └── generate_migration.py  # Migration script generation notebook
├── tests/
│  ├── conftest.py       # Shared pytest fixtures (SparkSession, sample schemas)
│  ├── test_schema_detector.py # Detector unit tests
│  ├── test_compatibility.py  # Compatibility checker tests
│  ├── test_schema_diff.py   # Schema diff tests
│  ├── test_migration_generator.py # Migration generator tests
│  └── test_impact_analyzer.py # Impact analyzer tests
└── guides/
  └── schema-evolution-strategy.md # Versioning, migration & impact analysis guide

Architecture

 ┌──────────────┐   ┌──────────────────┐   ┌─────────────────┐
 │ Upstream   │────▶│ Schema Detector │────▶│ Drift Report  │
 │ Data Source │   └──────────────────┘   └────────┬────────┘
 └──────────────┘                    │
                             ▼
            ┌──────────────────┐   ┌─────────────────┐
            │ Schema Diff   │◀────│ Compare v1/v2 │
            │ (field-level)  │   └─────────────────┘
            └───────┬──────────┘
                │
          ┌────────────┼────────────────┐
          ▼      │        ▼
     ┌────────────────┐  │   ┌───────────────────┐
     │ Compatibility │  │   │ Impact Analyzer  │
     │ Checker    │  │   │ (downstream deps) │
     └───────┬────────┘  │   └───────┬───────────┘
         │      │       │
     Compatible?   ┌─────▼─────┐   Risk
    ┌─────────┐   │ Migration │  Score OK?
    ▼ YES   ▼ NO  │ Generator │  ┌───────┐
 ┌──────────┐ ┌──────┐ └─────┬─────┘  ▼ YES  ▼ NO
 │ Generate │ │Alert/│    │  ┌──────────┐ ┌─────────┐
 │ Scripts │ │Block │    │  │ Apply  │ │ Manual │
 └─────┬────┘ └──────┘    │  │ Scripts │ │ Review │
    │            │  └─────┬────┘ └─────────┘
    ▼            │     │
┌──────────────┐        │     │
│ Schema    │◀──────────────┘──────────┘
│ Registry   │
│ (store ver.) │
└──────────────┘

Getting Started

Follow this guide to get schema evolution toolkit up and running in your environment.

Getting Started

1. Detect Schema Drift

from schema_detector import SchemaDetector

detector = SchemaDetector(spark)

# Compare a live table against an expected JSON schema
report = detector.compare_table_to_schema(
  table_name="catalog.bronze.customers",
  expected_schema_path="/Volumes/schemas/v2_customer.json",
)

print(report.summary())
# SchemaReport: 2 added columns, 1 type change, 0 removed columns

2. Detailed Schema Diff (No Spark Required)

from schema_diff import SchemaDiff

differ = SchemaDiff()
diff = differ.diff_from_files("v1_customer.json", "v2_customer.json")

# Print a categorized report
print(diff.to_text())

# Get Markdown for PR comments
print(diff.to_markdown())

# Check for breaking changes
if diff.has_breaking_changes:
  for bc in diff.breaking_changes:
    print(f" BREAKING: {bc.field_path} — {bc.message}")

3. Check Compatibility

from compatibility_checker import CompatibilityChecker

checker = CompatibilityChecker()

result = checker.check_backward_compatible(old_schema, new_schema)
if not result.is_compatible:
  for issue in result.issues:
    print(f" BREAKING: {issue}")

4. Generate Migration Scripts

from migration_generator import MigrationGenerator

gen = MigrationGenerator()
migration = gen.generate_from_files(
  "v1_customer.json", "v2_customer.json",
  table_name="catalog.silver.customers",
)

# Print forward SQL
print(migration.render_forward_sql())

# Print rollback SQL
print(migration.render_rollback_sql())

# Generate a Python migration module
print(migration.render_python_migration())

# Get structured JSON for CI/CD
print(migration.to_json())

5. Analyze Downstream Impact

from impact_analyzer import ImpactAnalyzer

analyzer = ImpactAnalyzer()
analyzer.load_manifest_file("dependency_manifest.json")
impact = analyzer.analyze_from_diff(diff, "catalog.silver.customers")

print(impact.summary())
# Impact Analysis: 3/5 assets impacted, risk_score=0.45, critical=1

if impact.critical_impacts:
  print("BLOCKED: SLA-critical assets affected!")

6. Apply a Migration

from schema_migrator import SchemaMigrator

migrator = SchemaMigrator(spark)

migrator.add_columns(
  table_name="catalog.silver.customers",
  columns={"loyalty_tier": "STRING", "last_login": "TIMESTAMP"},
)

7. Register a Schema Version

from schema_registry import SchemaRegistry

registry = SchemaRegistry(spark, registry_table="catalog.meta.schema_registry")
registry.register(
  subject="customers",
  version=2,
  schema=new_schema,
  description="Added loyalty_tier and last_login columns",
)

8. Validate at Runtime

from schema_validator import SchemaValidator

validator = SchemaValidator(spark)
result = validator.validate(df, expected_schema, mode="strict")
if not result.is_valid:
  raise ValueError(f"Schema validation failed: {result.errors}")

Schema Version Examples

The toolkit includes example schemas demonstrating common evolution patterns:

Customer Domain (v1 → v2 → v3)

Order Domain (v1 → v2)

Requirements

• Python 3.9+
• Schema Diff, Migration Generator, Impact Analyzer: Python stdlib only (no Spark required)
• Schema Detector, Migrator, Registry, Validator: PySpark 3.4+ / Databricks Runtime 13.3+
• Schema Registry: Delta Lake (delta-spark)
• Policy config loading: PyYAML