Delta Lake Versioning with Apache Spark: Mastering Data Evolution and Recovery

The Importance of Versioning in Data Lakes

Data lakes store vast, diverse datasets in formats like Parquet or JSON, often on distributed storage systems such as S3 or HDFS. Spark excels at processing these lakes, powering batch, streaming, and machine learning workloads. However, traditional data lakes lack mechanisms to track changes, making it hard to:

• Audit Modifications: Understand who changed what and when.
• Recover from Errors: Revert accidental deletes or corruptions.
• Reproduce Analyses: Query data as it existed at a specific point.
• Ensure Compliance: Meet regulatory requirements for data history.

Delta Lake addresses these challenges by adding a transactional layer to data lakes, with versioning as a core feature. Versioning records every change to a Delta table, enabling time travel—querying or restoring past states. This transforms data lakes into reliable, auditable platforms, seamlessly integrated with Spark’s DataFrame API. For a Delta Lake primer, see Spark Delta Lake guide.

What is Delta Lake Versioning?

Delta Lake versioning, often referred to as time travel, allows you to:

• Query Historical Data: Access a table’s state at a specific version or timestamp.
• Track Changes: View the history of operations (inserts, updates, deletes).
• Revert Changes: Restore a table to a previous version.
• Audit Operations: Understand how data evolved over time.

Versioning is powered by Delta Lake’s transaction log, which records all table modifications, ensuring ACID compliance and enabling seamless navigation through data states (Spark Delta Lake ACID transactions).

How Versioning Works

Delta Lake stores data in Parquet files, augmented by a transaction log in the _delta_log directory. The log consists of JSON files that capture:

• Operations: Inserts, updates, deletes, merges, or schema changes.
• Metadata: Table schema, partitioning, and file references.
• Version Numbers: Sequential IDs (0, 1, 2, …) for each committed transaction.
• Timestamps: When each operation occurred.

When you modify a Delta table (e.g., append data), Delta Lake:

1. Creates new Parquet files for added or updated data.
2. Records the operation in a new log entry, incrementing the version number.
3. Updates metadata to point to the current set of valid files.

To query a past version:

• Spark reads the log entry for the specified version or timestamp.
• It reconstructs the table’s state by referencing only the Parquet files valid at that point.
• No data is duplicated; versioning uses the log to filter files efficiently.

This enables operations like auditing, rollback, or historical analysis without overhead.

Practical Applications of Versioning

Versioning supports a range of use cases:

• Error Recovery: Revert accidental deletes or updates Spark Delta Lake rollback using time travel.
• Auditing: Track changes for compliance or debugging.
• Reproducible Analytics: Query data as it was during past analyses.
• Data Science: Experiment with datasets while preserving original states.
• Testing: Validate pipeline changes against historical data.

Versioning is automatic with Delta Lake, requiring no extra setup beyond creating a Delta table.

Setting Up Delta Lake with Spark

Let’s configure an environment to explore versioning with Spark and Delta Lake.

Prerequisites

1. Spark Installation:
   
   • Use Spark 3.5.x or later PySpark installation.
   • Verify:

   • Copy codespark-shell

1. Delta Lake Dependency:
   
   • Include Delta Lake 3.2.0 (compatible with Spark 3.5).
   • For PySpark:

   • Copy codespark = SparkSession.builder \
              .appName("DeltaVersioning") \
              .config("spark.jars.packages", "io.delta:delta-spark_2.12:3.2.0") \
              .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
              .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
              .getOrCreate()

   • For Scala, add to SBT:

   • Copy codelibraryDependencies += "io.delta" %% "delta-spark" % "3.2.0"

1. Storage:
   
   • Use a local directory (e.g., /tmp/delta) or cloud storage like S3.
   • Ensure write access.

Building a Versioned Delta Lake Pipeline

We’ll create a Delta Lake pipeline that:

• Initializes a Delta table with sales data.
• Performs multiple operations (insert, update, delete, merge) to create versions.
• Queries historical versions using version numbers and timestamps.
• Demonstrates rollback and auditing.

Examples are provided in PySpark and Scala, with both batch and streaming operations.

PySpark Versioning Pipeline

This pipeline showcases versioning through a series of table modifications and queries.

Code

Copy codefrom pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType
from delta.tables import DeltaTable
import time

# Initialize Spark with Delta
spark = SparkSession.builder \
    .appName("DeltaVersioningPipeline") \
    .config("spark.jars.packages", "io.delta:delta-spark_2.12:3.2.0") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .config("spark.sql.shuffle.partitions", "4") \
    .getOrCreate()

# Define schema
schema = StructType([
    StructField("order_id", IntegerType(), nullable=False),
    StructField("amount", IntegerType(), nullable=False),
    StructField("region", StringType(), nullable=False),
    StructField("timestamp", TimestampType(), nullable=False)
])

# Create initial data (Version 0)
data_v0 = [
    (1, 100, "North", "2024-10-01T10:00:00"),
    (2, 200, "South", "2024-10-01T10:01:00")
]
df_v0 = spark.createDataFrame(data_v0, schema)

# Write initial Delta table
table_path = "/tmp/delta/sales_versioned"
df_v0.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save(table_path)

print("Version 0:")
spark.read.format("delta").load(table_path).show(truncate=False)

# Update (Version 1)
delta_table = DeltaTable.forPath(spark, table_path)
delta_table.update(
    condition=col("order_id") == 1,
    set={"amount": lit(150)}
)
print("Version 1 (After Update):")
spark.read.format("delta").load(table_path).show(truncate=False)

# Delete (Version 2)
delta_table.delete(col("region") == "South")
print("Version 2 (After Delete):")
spark.read.format("delta").load(table_path).show(truncate=False)

# Merge (Version 3)
updates = [
    (1, 175, "North", "2024-10-01T10:03:00"),
    (3, 300, "West", "2024-10-01T10:04:00")
]
updates_df = spark.createDataFrame(updates, schema)
delta_table.alias("target").merge(
    updates_df.alias("source"),
    "target.order_id = source.order_id"
).whenMatchedUpdate(set={"amount": col("source.amount"), "timestamp": col("source.timestamp")}) \
 .whenNotMatchedInsertAll() \
 .execute()
print("Version 3 (After Merge):")
spark.read.format("delta").load(table_path).show(truncate=False)

# Query historical versions
print("Query Version 0:")
spark.read.format("delta").option("versionAsOf", 0).load(table_path).show(truncate=False)

# Query by timestamp (approximate, after Version 1)
print("Query by Timestamp (post-Version 1):")
spark.read.format("delta").option("timestampAsOf", "2024-10-01 10:10:00").load(table_path).show(truncate=False)

# View table history
print("Table History:")
delta_table.history().select("version", "timestamp", "operation", "operationParameters").show(truncate=False)

# Rollback to Version 0
spark.read.format("delta").option("versionAsOf", 0).load(table_path) \
    .write.format("delta").mode("overwrite").save(table_path)
print("After Rollback to Version 0:")
spark.read.format("delta").load(table_path).show(truncate=False)

# Streaming append (creates new versions)
new_data = [(4, 400, "East", "2024-10-01T10:05:00")]
new_df = spark.createDataFrame(new_data, schema)
new_df.write.format("delta").mode("append").save(table_path)

# Stream updates to console
streaming_df = spark.readStream.format("delta").load(table_path)
query = streaming_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .option("truncate", "false") \
    .trigger(processingTime="10 seconds") \
    .start()

# Simulate another append
more_data = [(5, 500, "North", "2024-10-01T10:06:00")]
more_df = spark.createDataFrame(more_data, schema)
more_df.write.format("delta").mode("append").save(table_path)

# Run streaming for 30 seconds
query.awaitTermination(30)
query.stop()

# Clean up
spark.stop()

Parameters Explained

1. Spark Session:
   
   • appName: Identifies the job in the UI.
   • spark.jars.packages: Adds Delta Lake dependency.
   • spark.sql.extensions: Enables Delta features.
   • spark.sql.catalog.spark_catalog: Configures Delta catalog.
   • spark.sql.shuffle.partitions: Limits partitions for small-scale testing Spark SQL shuffle partitions.

1. Schema:
   
   • Defines order_id, amount, region, and timestamp with nullable=False for strict validation.
   • TimestampType supports versioning and time-based queries.

1. Write Delta Table:
   
   • format("delta"): Specifies Delta format.
   • mode("overwrite"): Replaces existing data.
   • option("overwriteSchema", "true"): Allows schema updates.
   • save(table_path): Writes to /tmp/delta/sales_versioned.

1. Update:
   
   • DeltaTable.forPath: Loads the table.
   • update(condition, set): Modifies rows, creating a new version Spark DataFrame update.

1. Delete:
   
   • delete(condition): Removes rows, incrementing the version.

1. Merge:
   
   • merge(source, condition): Upserts data, creating another version.
   • whenMatchedUpdate: Updates matching rows.
   • whenNotMatchedInsertAll: Inserts new rows.

1. Time Travel:
   
   • option("versionAsOf", 0): Queries a specific version.
   • option("timestampAsOf", "2024-10-01 10:10:00"): Queries by approximate timestamp Spark time travel in Spark Delta Lake.

1. History:
   
   • history(): Retrieves operation metadata, including version, timestamp, and parameters.

1. Rollback:
   
   • Overwrites the table with a past version’s data, creating a new version.

1. Streaming:
   • readStream: Reads Delta table updates as a stream.
   • writeStream: Outputs new rows to the console.
   • outputMode("append"): Shows appended data.
   • trigger(processingTime="10 seconds"): Processes every 10 seconds PySpark streaming triggers.

Output

• Version 0:

• Copy code+--------+------+------+--------------------+
    |order_id|amount|region|timestamp           |
    +--------+------+------+--------------------+
    |1       |100   |North |2024-10-01 10:00:00|
    |2       |200   |South |2024-10-01 10:01:00|
    +--------+------+------+--------------------+

• Version 1 (After Update):

• Copy code+--------+------+------+--------------------+
    |order_id|amount|region|timestamp           |
    +--------+------+------+--------------------+
    |1       |150   |North |2024-10-01 10:00:00|
    |2       |200   |South |2024-10-01 10:01:00|
    +--------+------+------+--------------------+

• Version 2 (After Delete):

• Copy code+--------+------+------+--------------------+
    |order_id|amount|region|timestamp           |
    +--------+------+------+--------------------+
    |1       |150   |North |2024-10-01 10:00:00|
    +--------+------+------+--------------------+

• Version 3 (After Merge):

• Copy code+--------+------+------+--------------------+
    |order_id|amount|region|timestamp           |
    +--------+------+------+--------------------+
    |1       |175   |North |2024-10-01 10:03:00|
    |3       |300   |West  |2024-10-01 10:04:00|
    +--------+------+------+--------------------+

• Query Version 0:

• Copy code+--------+------+------+--------------------+
    |order_id|amount|region|timestamp           |
    +--------+------+------+--------------------+
    |1       |100   |North |2024-10-01 10:00:00|
    |2       |200   |South |2024-10-01 10:01:00|
    +--------+------+------+--------------------+

• Query by Timestamp (post-Version 1):

• Copy code+--------+------+------+--------------------+
    |order_id|amount|region|timestamp           |
    +--------+------+------+--------------------+
    |1       |150   |North |2024-10-01 10:00:00|
    |2       |200   |South |2024-10-01 10:01:00|
    +--------+------+------+--------------------+

• Table History (example, simplified):

• Copy code+-------+-------------------+---------+--------------------+
    |version|timestamp          |operation|operationParameters |
    +-------+-------------------+---------+--------------------+
    |0      |2024-10-01 10:00:00|WRITE    |{mode=overwrite}   |
    |1      |2024-10-01 10:00:05|UPDATE   |{predicate=order_id=1}|
    |2      |2024-10-01 10:00:10|DELETE   |{predicate=region='South'}|
    |3      |2024-10-01 10:00:15|MERGE    |{predicate=order_id}|
    +-------+-------------------+---------+--------------------+

• After Rollback to Version 0:

• Copy code+--------+------+------+--------------------+
    |order_id|amount|region|timestamp           |
    +--------+------+------+--------------------+
    |1       |100   |North |2024-10-01 10:00:00|
    |2       |200   |South |2024-10-01 10:01:00|
    +--------+------+------+--------------------+

• Streaming Output (after appends):

• Copy code-------------------------------------------
    Batch: 1
    -------------------------------------------
    |order_id|amount|region|timestamp           |
    |4       |400   |East  |2024-10-01 10:05:00|
  
    -------------------------------------------
    Batch: 2
    -------------------------------------------
    |order_id|amount|region|timestamp           |
    |5       |500   |North |2024-10-01 10:06:00|

Scala Versioning Pipeline

The same pipeline in Scala:

Code

Copy codeimport org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, lit}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType, TimestampType}
import io.delta.tables.DeltaTable
import org.apache.spark.sql.streaming.Trigger

object DeltaVersioningPipeline {
  def main(args: Array[String]): Unit = {
    // Initialize Spark
    val spark = SparkSession.builder()
      .appName("DeltaVersioningPipeline")
      .config("spark.jars.packages", "io.delta:delta-spark_2.12:3.2.0")
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
      .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
      .config("spark.sql.shuffle.partitions", "4")
      .getOrCreate()

    import spark.implicits._

    // Define schema
    val schema = StructType(Seq(
      StructField("order_id", IntegerType, nullable = false),
      StructField("amount", IntegerType, nullable = false),
      StructField("region", StringType, nullable = false),
      StructField("timestamp", TimestampType, nullable = false)
    ))

    // Create initial data (Version 0)
    val dataV0 = Seq(
      (1, 100, "North", "2024-10-01T10:00:00"),
      (2, 200, "South", "2024-10-01T10:01:00")
    )
    val dfV0 = dataV0.toDF("order_id", "amount", "region", "timestamp").select(
      $"order_id".cast(IntegerType),
      $"amount".cast(IntegerType),
      $"region".cast(StringType),
      $"timestamp".cast(TimestampType)
    )

    // Write initial Delta table
    val tablePath = "/tmp/delta/sales_versioned"
    dfV0.write.format("delta")
      .mode("overwrite")
      .option("overwriteSchema", "true")
      .save(tablePath)

    println("Version 0:")
    spark.read.format("delta").load(tablePath).show(truncate = false)

    // Update (Version 1)
    val deltaTable = DeltaTable.forPath(spark, tablePath)
    deltaTable.update(
      condition = col("order_id") === 1,
      set = Map("amount" -> lit(150))
    )
    println("Version 1 (After Update):")
    spark.read.format("delta").load(tablePath).show(truncate = false)

    // Delete (Version 2)
    deltaTable.delete(col("region") === "South")
    println("Version 2 (After Delete):")
    spark.read.format("delta").load(tablePath).show(truncate = false)

    // Merge (Version 3)
    val updates = Seq(
      (1, 175, "North", "2024-10-01T10:03:00"),
      (3, 300, "West", "2024-10-01T10:04:00")
    )
    val updatesDf = updates.toDF("order_id", "amount", "region", "timestamp").select(
      $"order_id".cast(IntegerType),
      $"amount".cast(IntegerType),
      $"region".cast(StringType),
      $"timestamp".cast(TimestampType)
    )
    deltaTable.alias("target")
      .merge(updatesDf.alias("source"), "target.order_id = source.order_id")
      .whenMatched.update(set = Map(
        "amount" -> col("source.amount"),
        "timestamp" -> col("source.timestamp")
      ))
      .whenNotMatched.insertAll()
      .execute()
    println("Version 3 (After Merge):")
    spark.read.format("delta").load(tablePath).show(truncate = false)

    // Query historical versions
    println("Query Version 0:")
    spark.read.format("delta").option("versionAsOf", 0).load(tablePath).show(truncate = false)

    // Query by timestamp
    println("Query by Timestamp (post-Version 1):")
    spark.read.format("delta").option("timestampAsOf", "2024-10-01 10:10:00").load(tablePath).show(truncate = false)

    // View table history
    println("Table History:")
    deltaTable.history().select("version", "timestamp", "operation", "operationParameters").show(truncate = false)

    // Rollback to Version 0
    spark.read.format("delta").option("versionAsOf", 0).load(tablePath)
      .write.format("delta").mode("overwrite").save(tablePath)
    println("After Rollback to Version 0:")
    spark.read.format("delta").load(tablePath).show(truncate = false)

    // Streaming append
    val newData = Seq((4, 400, "East", "2024-10-01T10:05:00"))
    val newDf = newData.toDF("order_id", "amount", "region", "timestamp").select(
      $"order_id".cast(IntegerType),
      $"amount".cast(IntegerType),
      $"region".cast(StringType),
      $"timestamp".cast(TimestampType)
    )
    newDf.write.format("delta").mode("append").save(tablePath)

    // Stream updates
    val streamingDf = spark.readStream.format("delta").load(tablePath)
    val query = streamingDf.writeStream
      .outputMode("append")
      .format("console")
      .option("truncate", "false")
      .trigger(Trigger.ProcessingTime("10 seconds"))
      .start()

    // Simulate another append
    val moreData = Seq((5, 500, "North", "2024-10-01T10:06:00"))
    val moreDf = moreData.toDF("order_id", "amount", "region", "timestamp").select(
      $"order_id".cast(IntegerType),
      $"amount".cast(IntegerType),
      $"region".cast(StringType),
      $"timestamp".cast(TimestampType)
    )
    moreDf.write.format("delta").mode("append").save(tablePath)

    // Run for 30 seconds
    query.awaitTermination(30)
    query.stop()

    // Clean up
    spark.stop()
  }
}

Running the Scala Application

1. Package with SBT:

Copy codesbt package

1. Submit:

Copy codespark-submit --class DeltaVersioningPipeline \
       --packages io.delta:delta-spark_2.12:3.2.0 \
       target/scala-2.12/your-app.jar

The output matches the PySpark example, demonstrating versioning’s versatility.

Alternative Approach: SQL for Versioning

Delta Lake supports SQL for versioning, offering a familiar interface for database users.

PySpark SQL Example

Copy code# Create table
spark.sql("""
    CREATE TABLE delta_sales (
        order_id INT,
        amount INT,
        region STRING,
        timestamp TIMESTAMP
    ) USING delta
    LOCATION '/tmp/delta/sales_versioned'
""")

# Insert initial data (Version 0)
spark.sql("""
    INSERT INTO delta_sales
    VALUES
        (1, 100, 'North', '2024-10-01T10:00:00'),
        (2, 200, 'South', '2024-10-01T10:01:00')
""")
print("Version 0:")
spark.sql("SELECT * FROM delta_sales").show(truncate=False)

# Update (Version 1)
spark.sql("UPDATE delta_sales SET amount = 150 WHERE order_id = 1")
print("Version 1:")
spark.sql("SELECT * FROM delta_sales").show(truncate=False)

# Query history
print("Table History:")
spark.sql("DESCRIBE HISTORY delta_sales").select("version", "timestamp", "operation").show(truncate=False)

# Time travel
print("Version 0:")
spark.sql("SELECT * FROM delta_sales VERSION AS OF 0").show(truncate=False)

This SQL-based approach achieves similar versioning functionality (PySpark SQL introduction).

Best Practices

Leverage versioning effectively with these tips:

• Use Version Queries for Auditing: Regularly check history() to track changes.
• Test Rollbacks: Practice restoring versions in a test environment Spark Delta Lake rollback using time travel.
• Combine with Checkpointing: Ensure streaming reliability PySpark streaming checkpointing.
• Optimize Tables: Compact files to manage log size:

• Copy codedelta_table.optimize().executeCompaction()

• Monitor Performance: Use the Spark UI Spark how to debug Spark applications.
• Document Changes: Note version numbers for critical updates.

Common Pitfalls

Avoid these mistakes:

• Ignoring History: Misses auditing opportunities. Solution: Review history().
• Excessive Versions: Bloats logs. Solution: Run VACUUM to clean old versions:

• Copy codedelta_table.vacuum(168) # Retain 7 days

• No Checkpointing: Risks streaming failures. Solution: Set checkpointLocation.
• Incorrect Timestamps: Skews time-based queries. Solution: Validate timestampAsOf.

Monitoring and Validation

Ensure versioning works correctly:

• Spark UI: Track job metrics and resource usage.
• Table History: Verify operations with history().
• Version Queries: Confirm expected data states.
• Logs: Check for errors PySpark logging.

Next Steps

Continue exploring Delta Lake with:

• Time travel workflows Spark working with time travel.
• Streaming with Delta PySpark streaming input sources.
• Optimization techniques Spark how to optimize jobs for max performance.

Try the Databricks Community Edition for hands-on practice.

By mastering Delta Lake versioning, you’ll ensure your data lakes are auditable, recoverable, and ready for scalable analytics with Apache Spark.