Skip to content

Delta Lake

Delta Lake is an open-source storage layer that brings ACID transactions, schema evolution, and time travel to Apache Spark. This sheet covers all core operations.

Table of Contents

Installation / Setup

# Delta Lake is included in Databricks
# For standalone Spark, install:
# pip install delta-spark

# SparkSession with Delta
from pyspark.sql import SparkSession

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

Core Syntax

Read and Write Delta

# Read Delta table
df = spark.read.format("delta").load("/path/to/delta/table")
df = spark.table("database.table_name")

# Write Delta table
df.write.format("delta").save("/path/to/delta/table")
df.write.format("delta").saveAsTable("database.table_name")

# Write modes
df.write.format("delta").mode("overwrite").save("/path/to/table")
df.write.format("delta").mode("append").save("/path/to/table")
df.write.format("delta").mode("ignore").save("/path/to/table")

Schema Evolution

# Add new column (mergeSchema option)
new_df = df.withColumn("new_col", F.lit("value"))
new_df.write.format("delta").mode("append") \
    .option("mergeSchema", "true") \
    .save("/path/to/table")

# Overwrite schema (use with caution)
df.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("/path/to/table")

Common Operations

Time Travel

# Read by version number
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("/path/to/table")
df_v1 = spark.read.format("delta").option("versionAsOf", 1).load("/path/to/table")

# Read by timestamp
df_before = spark.read.format("delta") \
    .option("timestampAsOf", "2026-01-01") \
    .load("/path/to/table")

# View history
delta_table = DeltaTable.forPath(spark, "/path/to/table")
history = delta_table.history()
history.show()

# Restore to previous version
delta_table.restoreToVersion(0)

MERGE (Upsert)

from delta.tables import DeltaTable

# Load target table
delta_table = DeltaTable.forPath(spark, "/path/to/table")

# Prepare updates
updates_df = spark.read.format("delta").load("/path/to/updates")

# MERGE operation
delta_table.alias("target").merge(
    updates_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(
    set={
        "target.name": "source.name",
        "target.amount": "source.amount",
        "target.updated_at": "source.updated_at"
    }
).whenNotMatchedInsert(
    values={
        "id": "source.id",
        "name": "source.name",
        "amount": "source.amount",
        "created_at": "source.created_at",
        "updated_at": "source.updated_at"
    }
).execute()

# Conditional merge
delta_table.alias("target").merge(
    updates_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(
    condition="source.amount > target.amount",
    set={"target.amount": "source.amount"}
).whenNotMatchedInsertAll().execute()

DELETE

# Delete rows
delta_table.delete("status = 'inactive'")

# Delete with complex condition
delta_table.delete(F.col("created_at") < "2025-01-01")

UPDATE

# Update rows
delta_table.update(
    condition="status = 'pending'",
    set={"status": F.lit("active")}
)

Optimization

# Compact small files
delta_table.optimize().executeCompaction()

# Z-order by column (optimize for queries)
delta_table.optimize().executeZOrderBy("date", "region")

# Remove old files
delta_table.vacuum()  # default: keep 7 days
delta_table.vacuum(168)  # keep 168 hours (7 days)

# Auto-optimize on write
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

Data Quality Constraints

# Add constraints
spark.sql("""
    ALTER TABLE database.table_name
    ADD CONSTRAINT valid_amount CHECK (amount > 0)
""")

spark.sql("""
    ALTER TABLE database.table_name
    ADD CONSTRAINT valid_email CHECK (email LIKE '%@%.%')
""")

# View constraints
spark.sql("DESCRIBE TABLE EXTENDED database.table_name").show(truncate=False)

Practical Examples

Example 1 — Full ETL with Delta

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

# Read raw data
raw_df = spark.read.csv("dbfs:/raw/events/", header=True, inferSchema=True)

# Clean
clean_df = raw_df \
    .dropDuplicates(["id"]) \
    .filter(F.col("amount") > 0) \
    .withColumn("date", F.to_date("timestamp")) \
    .withColumn("processed_at", F.current_timestamp())

# Write to Delta with partitioning
clean_df.write.format("delta") \
    .mode("overwrite") \
    .partitionBy("date") \
    .saveAsTable("events_clean")

Example 2 — Incremental merge

# Read new data
new_data = spark.read.format("delta").load("dbfs:/staging/events_new/")

# Merge into target
delta_table = DeltaTable.forPath(spark, "dbfs:/data/events")

delta_table.alias("target").merge(
    new_data.alias("source"),
    "target.id = source.id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# Optimize after merge
delta_table.optimize().executeCompaction()

Example 3 — Data quality pipeline

# Load table
df = spark.table("events")

# Validate
invalid = df.filter(~F.col("email").rlike("^[^@]+@[^@]+\\.[^@]+$"))

if invalid.count() > 0:
    print(f"Found {invalid.count()} invalid records")
    invalid.write.format("delta").mode("append").saveAsTable("events_quarantine")
else:
    print("All records valid")

Common Patterns

Pattern: schema evolution

# Add new column incrementally
df_with_new_col = df.withColumn("new_metric", F.col("value") * 1.1)

# Append with schema evolution
df_with_new_col.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("events")

Pattern: audit logging

# Track changes with append
df.write.format("delta") \
    .mode("append") \
    .saveAsTable("events_audit")

# Query history
spark.sql("""
    SELECT * FROM tableHistory('events_audit')
    ORDER BY version DESC
""").show()

Pattern: multi-hop pipeline

# Bronze: raw data
raw_df.write.format("delta").mode("append").saveAsTable("bronze")

# Silver: cleaned
spark.table("bronze") \
    .dropDuplicates() \
    .filter(F.col("amount") > 0) \
    .write.format("delta").mode("overwrite").saveAsTable("silver")

# Gold: aggregated
spark.table("silver") \
    .groupBy("date", "region") \
    .agg(F.sum("amount").alias("revenue")) \
    .write.format("delta").mode("overwrite").saveAsTable("gold")

Best Practices

  • Use Delta Lake for all production data — ACID transactions prevent corruption.
  • Partition by date — optimize queries and enable efficient pruning.
  • Run OPTIMIZE regularly — compact small files for better performance.
  • Use Z-ordering — for frequently filtered columns.
  • Set up data quality constraints — prevent bad data from entering.
  • Keep change history — enables debugging and audit.
  • Use time travel — for recovery and reproducibility.
  • VACUUM old files — control storage costs.

Common Mistakes

Mistake Fix
Not using Delta format Use Delta for all production tables
Over-partitioning Partition by low-cardinality columns only
Not running OPTIMIZE Schedule regular optimization
Not setting data quality constraints Add constraints for critical columns
Ignoring small file problem Use auto-compaction or manual OPTIMIZE
Not using VACUUM Run VACUUM to clean old files

Troubleshooting

Symptom Cause / Fix
Slow queries Run OPTIMIZE; check partition strategy
File not found Check path; verify table exists
Schema mismatch Use mergeSchema option
Too many small files Run OPTIMIZE and VACUUM
Out of memory Increase cluster size; repartition
Version not found Check history; table may have been vacuumed

Official Documentation

Last Updated

2026-09-06 — Initial version.