Question

Explain how Delta Lake provides ACID transactions on Spark. How does the MERGE operation handle updates and deletes? What is the OPTIMIZE command and how does Z-ORDER improve query performance? Describe time travel and its use cases for data governance.

Detailed Answer

1. Delta Lake Architecture

Transaction log contains:

1. Metadata (schema, partition columns)

2. Protocol (reader/writer versions)

3. Add file actions (new data files)

4. Remove file actions (deleted files)

5. Protocol version (feature compatibility)

from delta import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .appName("DeltaLakeDeepDive") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", 
            "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Create Delta table
df = spark.range(1000000).withColumn(
    "value", F.randn()
).withColumn(
    "date", F.date_add(F.lit("2024-01-01"), (F.rand() * 365).cast("int"))
)

# Write as Delta table
df.write.format("delta").mode("overwrite").save("/delta/events")

# Read Delta table
delta_df = spark.read.format("delta").load("/delta/events")

2. ACID Transactions

# Delta Lake ACID properties:
# Atomicity: Each commit is all-or-nothing
# Consistency: Schema enforcement, constraints
# Isolation: Serializable isolation for writers
# Durability: Commit log persisted to storage

# How ACID works:
# 1. Writer acquires commit lock
# 2. Writer creates data files
# 3. Writer writes commit JSON to _delta_log/
# 4. Commit JSON contains all file operations
# 5. Other writers see committed state

# Atomic write example:
delta_table = DeltaTable.forPath(spark, "/delta/events")

# Atomic overwrite
delta_table.alias("target").merge(
    source_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# Transaction log structure:
# Commit 0000:
# {
#   "add": {
#     "path": "part-00000.parquet",
#     "size": 1024000,
#     "modificationTime": 1704067200000,
#     "dataChange": true,
#     "stats": "{\"numRecords\":100000}"
#   }
# }
#
# Commit 0001:
# {
#   "remove": {
#     "path": "part-00000.parquet",
#     "deletionTimestamp": 1704067260000,
#     "dataChange": true
#   }
# }
# {
#   "add": {
#     "path": "part-00000-updated.parquet",
#     "size": 1024000,
#     "modificationTime": 1704067260000,
#     "dataChange": true
#   }
# }

3. MERGE Operation

# MERGE: Upsert operation (UPDATE + INSERT)
# Handles: updates, inserts, deletes, conditional operations

# Basic MERGE:
delta_table.alias("target").merge(
    source_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(
    condition="source.timestamp > target.timestamp",
    set={
        "value": "source.value",
        "timestamp": "source.timestamp"
    }
).whenNotMatchedInsert(
    values={
        "id": "source.id",
        "value": "source.value",
        "timestamp": "source.timestamp"
    }
).execute()

# Advanced MERGE with multiple conditions:
delta_table.alias("target").merge(
    source_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(
    condition="source.operation = 'UPDATE'",
    set={
        "value": "source.value",
        "updated_at": "current_timestamp()"
    }
).whenMatchedDelete(
    condition="source.operation = 'DELETE'"
).whenNotMatchedInsert(
    condition="source.operation != 'DELETE'",
    values={
        "id": "source.id",
        "value": "source.value",
        "created_at": "current_timestamp()"
    }
).execute()

# MERGE performance optimization:
# 1. Partition pruning: MERGE respects partition columns
# 2. Data skipping: Uses file statistics to skip files
# 3. Lazy execution: Files only rewritten if matched

# MERGE metrics:
# Source rows: 1,000,000
# Target files: 100
# Files matched: 50 (contain matching keys)
# Files rewritten: 20 (actually changed)
# Time: 45 seconds (vs hours for full rewrite)

4. OPTIMIZE and Z-ORDER

# OPTIMIZE: Compacts small files into larger ones
# Reduces file count, improves query performance

# Basic OPTIMIZE
delta_table.optimize().execute()

# OPTIMIZE with Z-ORDER
delta_table.optimize().executeZOrderBy("user_id", "event_type")

# Z-ORDER: Multi-dimensional clustering
# Files are organized by multiple columns simultaneously
# Improves data skipping for queries filtering on any Z-ORDER column

# Z-ORDER algorithm:
# 1. For each file, compute Z-value = interleave bits of column values
# 2. Sort files by Z-value
# 3. Write files in Z-value order
#
# Example with 2 columns (user_id, event_type):
# user_id=1, event_type="click": Z = interleave(1, "click")
# user_id=2, event_type="view":  Z = interleave(2, "view")
# Files are written in Z-value order

# Data skipping with Z-ORDER:
# Query: WHERE user_id = 123 AND event_type = 'click'
# Without Z-ORDER: Scan all 100 files
# With Z-ORDER: Scan only 5 files (0.5% of data)

# OPTIMIZE scheduling:
# Run OPTIMIZE during off-peak hours
# Use Delta Lake OPTIMIZE command or Spark SQL

# Automatic compaction (Delta Lake 2.0+):
spark.conf.set("delta.autoOptimize.optimizeWrite", "true")
spark.conf.set("delta.autoOptimize.autoCompact", "true")

# OPTIMIZE metrics:
# Files before: 1000
# Files after: 50
# Data files removed: 950
# Data files added: 50
# Average file size: 128 MB (target)
# Total data: 6.4 GB

5. Time Travel

# Time Travel: Query historical versions of Delta table
# Uses transaction log to reconstruct past state

# Query by version number
current_df = spark.read.format("delta").load("/delta/events")
v5_df = spark.read.format("delta").option("versionAsOf", 5).load("/delta/events")

# Query by timestamp
ts_df = spark.read.format("delta") \
    .option("timestampAsOf", "2024-06-15") \
    .load("/delta/events")

# View history
history = delta_table.history()
history.show(truncate=False)
# +-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-------------+----------+-------------+----------+
# |version|timestamp          |userId|userName|operation|operationParameters|job |notebook|clusterId|readVersion|isolationLevel|operationMetrics|userMetadata|engineInfo|
# +-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-------------+----------+-------------+----------+
# |5      |2024-06-15 10:30:00|user1 |admin   |MERGE    |{predicate: [...]  |... |...     |...      |4          |WriteSerializable|{numTargetRowsInserted: 1000, ...}|null       |Apache-Spark-3.5.0|
# +-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-------------+----------+-------------+----------+

# Time travel use cases:
# 1. Debugging: Query data as it was before a bug
# 2. Auditing: Track data changes over time
# 3. Reprocessing: Re-run queries on historical data
# 4. Rollback: Restore to previous version

# Rollback to previous version:
delta_table.restoreToVersion(5)

# Time travel with SQL:
spark.sql("""
    SELECT * FROM delta.`/delta/events`
    VERSION AS OF 5
""").show()

spark.sql("""
    SELECT * FROM delta.`/delta/events`
    TIMESTAMP AS OF '2024-06-15'
""").show()

6. Schema Evolution

# Schema evolution: Add/modify columns over time
# Delta Lake supports safe schema changes

# Add column
spark.sql("""
    ALTER TABLE delta.`/delta/events`
    ADD COLUMNS (new_column STRING)
""")

# Rename column (Delta Lake 2.0+)
spark.sql("""
    ALTER TABLE delta.`/delta/events`
    RENAME COLUMN old_name TO new_name
""")

# Schema evolution during write
df_with_new_column = spark.range(1000).withColumn(
    "new_field", F.lit("value")
)

# mergeSchema allows adding new columns
df_with_new_column.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save("/delta/events")

# Overwrite schema (replaces entire schema)
df_new_schema.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("/delta/events")

# Schema enforcement:
# Delta Lake enforces schema on write
# Appending data with different schema raises error
# Unless mergeSchema option is enabled

7. Constraints and Data Quality

# Delta Lake constraints ensure data quality
# Constraints are checked on write and MERGE

# Add NOT NULL constraint
spark.sql("""
    ALTER TABLE delta.`/delta/events`
    ADD CONSTRAINT user_id_not_null CHECK (user_id IS NOT NULL)
""")

# Add CHECK constraint
spark.sql("""
    ALTER TABLE delta.`/delta/events`
    ADD CONSTRAINT amount_positive CHECK (amount >= 0)
""")

# Add FOREIGN KEY constraint (Delta Lake 2.1+)
spark.sql("""
    ALTER TABLE delta.`/delta/events`
    ADD CONSTRAINT fk_customer
    FOREIGN KEY (customer_id) REFERENCES delta.`/delta/customers`(id)
""")

# Constraint violation handling:
# 1. Write operation fails if constraint violated
# 2. Error message shows which constraint failed
# 3. No partial writes (atomicity)

# Data quality with Delta Live Tables (DLT):
# @dlt.table
# def events():
#     return spark.read.format("delta").load("/delta/events")
#
# @dlt.expect("amount >= 0", "Amount must be positive")
# @dlt.expect_or_drop("user_id IS NOT NULL")
# def clean_events():
#     return dlt.read("events")

8. Performance Tuning

# Delta Lake performance tuning:

# 1. File sizing
# Target: 128MB - 1GB per file
# OPTIMIZE targets: spark.databricks.delta.optimize.maxFileSize (1GB)

# 2. Data skipping
# Z-ORDER on frequently filtered columns
delta_table.optimize().executeZOrderBy("user_id", "event_type")

# 3. Partition pruning
# Partition by high-cardinality, frequently filtered columns
df.write.format("delta") \
    .partitionBy("date", "user_type") \
    .save("/delta/events")

# 4. Bloom filters (Delta Lake 2.4+)
delta_table.createOrReplaceTempView("events")
spark.sql("""
    ALTER TABLE delta.`/delta/events`
    SET TBLPROPERTIES (
        'delta.bloomFilter.columns' = 'user_id,event_type',
        'delta.bloomFilter.ndv' = '1000000'
    )
""")

# 5. Liquid clustering (Delta Lake 3.0+)
# Replaces partitioning and Z-ORDER with unified clustering
df.write.format("delta") \
    .option("delta.clusteringColumns", "user_id, event_type") \
    .save("/delta/events")

# 6. Compaction scheduling
# Run during off-peak hours
# Use Delta Lake APIs or Spark SQL

# Monitor Delta Lake metrics:
delta_table.detail().show(truncate=False)
# Shows: numFiles, sizeInBytes, numOutputBytes, etc.

Summary

Delta Lake transforms Spark into a reliable lakehouse platform with ACID transactions, schema evolution, and time travel for enterprise-grade data management.