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