hari-prasad
Valued Contributor II

‎01-09-2025 11:47 PM - edited ‎01-10-2025 12:14 AM

Hi @geckopher ,
To address your concerns about managing schema evolution, tracking metadata lineage, and efficiently updating schemas in Delta tables, here are some best practices and strategies:

Tracking Schema Changes

Delta Table History: Utilize Delta Lake's built-in history feature to track changes to your Delta tables. You can use the DESCRIBE HISTORY command to view the history of operations performed on a Delta table.

 

 

DESCRIBE HISTORY <delta.`/path/to/delta-table`>

 

 

 

Efficient Schema Updates

1. Schema Evolution: Use Delta Lak's schema evolution capabilities to automatically handle schema changes. When performing a MERGE operation, you can use the mergeSchema option to allow Delta Lake to auto merge the schema changes.

Note: The overwriteSchema option is used when you want to completely overwrite the schema of the target table with the schema of the source data.

 

 

deltaTable
    .alias("t")
    .merge(
        sourceDF.alias("s"),
        "s.key = t.key"
    )
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .option("mergeSchema", "true")
    .execute()

 

 

 

2. Conditional Schema Updates: Before performing a schema update, check if the new column already exists. This can be done programmatically by inspecting the schema of the Delta table.

 

 

from delta.tables import DeltaTable

deltaTable = DeltaTable.forPath(spark, "/path/to/delta-table")
schema = deltaTable.toDF().schema

if "new_column" not in schema.fieldNames():
    # Perform schema update
    deltaTable.updateExpr(
        condition="true",
        set={"new_column": "NULL"}
    )

 

 

 

 

Example Code

 

 

from delta.tables import DeltaTable
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SchemaManagement").getOrCreate()

# Path to Delta table
delta_table_path = "/path/to/delta-table"

# Check if the new column exists
deltaTable = DeltaTable.forPath(spark, delta_table_path)
schema = deltaTable.toDF().schema

if "new_column" not in schema.fieldNames():
    # Apply schema changes
    spark.sql(f"""
    ALTER TABLE delta.`{delta_table_path}`
    ADD COLUMNS (new_column STRING)
    """)

# Perform the MERGE operation
sourceDF = spark.read.format("delta").load("/path/to/source-data")
(
    deltaTable.alias("t")
    .merge(sourceDF.alias("s"), "s.key = t.key")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .option("mergeSchema", "true")
    .execute()
)

# Track changes
history = spark.sql(f"DESCRIBE HISTORY delta.`{delta_table_path}`")
display(history)

 

 

 
By following these best practices and strategies, you can manage schema evolution, track metadata lineage, and efficiently update schemas in Delta tables without unnecessary repetition or performance overhead.
 
Refer this link for more understanding Delta Lake Schema Evolution | Delta Lake
 
Regards,
Hai Prasad


Regards,
Hari Prasad
0 Kudos