Module 5: Delta Lake Fundamentals

• Delta Lake = open-source table format and storage layer
• Data stays in Parquet files. A transaction log adds reliability on top
• Default table format on Databricks. CREATE TABLE without a format gives a Delta table

• A half-finished write to a Parquet folder leaves partial files that readers see
• To change one row in Parquet you rewrite the folder yourself
• Parquet is fine for one-off exports and files you receive
• Everything you query repeatedly belongs in a Delta table

• Data files: Parquet, optionally with deletion vectors
• Log: ordered numbered commits (00000.json, 00001.json, ...), plus periodic checkpoints
• File statistics (min, max, null counts) in the log let Spark skip files that can't match a filter
• Managed tables: don't touch the files yourself. Use SQL or DataFrame APIs

%sql
DESCRIBE DETAIL orders_delta

• Look at format, location, numFiles, sizeInBytes

%sql
DESCRIBE HISTORY orders_delta

• One row per commit, newest first
• Key columns: version, timestamp, operation, operationParameters, operationMetrics
• First run: version 0 is the CREATE OR REPLACE TABLE AS SELECT from setup

• A commit lists files added and files removed. Nothing is edited in place
• Table history is kept for 30 days by default (delta.logRetentionDuration)

display(spark.sql("DESCRIBE HISTORY orders_delta").select(
    "version", "timestamp", "operation", "operationMetrics"))

• Atomic: a failed job leaves no half-written table
• Isolated: readers see the last committed version, not a write in progress
• Writers use optimistic concurrency: two writers touching the same files, the second commit can fail with a conflict exception and should retry
• Appends from different jobs rarely conflict. Overlapping updates and deletes can

# atomicity: this write fails part-way, so the table doesn't change
before = spark.table("orders_delta").count()
try:
    spark.sql("""
        INSERT INTO orders_delta
        SELECT order_id + 10000, customer_id, product_id,
               CASE WHEN order_id = 3 THEN raise_error('boom') ELSE quantity END,
               status, order_ts
        FROM orders WHERE order_id <= 5
    """)
except Exception as e:
    print("failed:", str(e).splitlines()[0])
print(before, spark.table("orders_delta").count())    # same number twice

• raise_error fails the write while it runs. The rows before it are not kept

• Databricks calls managed tables the default and recommended type
• Managed: DROP TABLE removes data after a recovery window (7 days by default). UNDROP TABLE works inside it
• External: dropping removes only the metadata. You delete the files
• Some features, like predictive optimization, only apply to managed tables
• External tables need a storage credential and external location (Module 6)

%sql
DESCRIBE EXTENDED orders_delta

• Look for Type (MANAGED) and Location

%sql
CREATE TABLE IF NOT EXISTS customers_delta (
  customer_id INT NOT NULL,
  name        STRING,
  country     STRING,
  signup_date DATE
)
COMMENT 'Practice customers'

# from a DataFrame
(spark.table("customers")
      .write.mode("overwrite")
      .saveAsTable("customers_delta"))

# from a query
spark.sql("CREATE OR REPLACE TABLE big_orders AS SELECT * FROM orders WHERE quantity >= 4")

• CREATE OR REPLACE TABLE keeps the table's history. DROP + CREATE doesn't
• NOT NULL and CHECK constraints are enforced on every write

%sql
ALTER TABLE orders_delta ADD CONSTRAINT qty_positive CHECK (quantity > 0 OR quantity IS NULL)

%sql
UPDATE orders_delta SET status = 'cancelled' WHERE order_id = 1

%sql
DELETE FROM orders_delta WHERE quantity IS NULL

• The 15 null-quantity rows are gone: count is now 1985

Upsert with MERGE:

%sql
CREATE OR REPLACE TEMP VIEW order_updates AS
SELECT order_id, customer_id, product_id, quantity, 'returned' AS status, order_ts
FROM orders_delta WHERE order_id <= 3
UNION ALL
SELECT 2001, 1, 1, 2, 'completed', current_timestamp()

%sql
MERGE INTO orders_delta AS t
USING order_updates AS s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

• Orders 1 to 3 are updated, order 2001 is inserted: count is 1986
• Optional clause: WHEN NOT MATCHED BY SOURCE THEN DELETE (or UPDATE) for target rows missing from the source

extra = spark.table("orders").limit(3).withColumn("note", F.lit("hi"))
try:
    extra.write.mode("append").saveAsTable("orders_delta")
except Exception as e:
    print("rejected:", str(e).splitlines()[0])

• Append with a new column is rejected by default: a typo can't silently add columns
• Opt in to evolution when the change is intended:

extra.write.mode("append").option("mergeSchema", "true").saveAsTable("orders_delta")
spark.table("orders_delta").printSchema()       # now has "note"

• Or evolve explicitly: ALTER TABLE orders_delta ADD COLUMN note STRING
• MERGE can evolve schema too, with automatic schema evolution enabled

%sql
DESCRIBE HISTORY orders_delta

%sql
SELECT COUNT(*) FROM orders_delta VERSION AS OF 0

• Version 0 still has all 2000 original rows (if you reran setup, pick the version with 2000 rows from the history)
• Same with a timestamp: TIMESTAMP AS OF '2026-10-19'
• In PySpark: spark.read.option("versionAsOf", 0).table("orders_delta")
• Short form: SELECT * FROM orders_delta@v0

Undo a mistake:

%sql
RESTORE TABLE orders_delta TO VERSION AS OF 0

• Needs MODIFY on the table. The table is back to 2000 rows
• RESTORE is itself a new commit: history isn't erased
• It fails if VACUUM already deleted the files that version needs

%sql
VACUUM orders_delta DRY RUN

• DRY RUN lists what would be deleted. Right after creating the table it lists nothing: files are younger than 7 days
• Default retention: 7 days for data files (delta.deletedFileRetentionDuration)
• Log (history) retention: 30 days (delta.logRetentionDuration)
• Predictive optimization runs VACUUM for managed tables

%sql
ALTER TABLE orders_delta SET TBLPROPERTIES (
  'delta.deletedFileRetentionDuration' = 'interval 14 days',
  'delta.logRetentionDuration'         = 'interval 30 days')

%sql
OPTIMIZE orders_delta

• Output shows files removed and added
• Each UPDATE, MERGE or small append can leave small files
• Fewer, larger files = fewer tasks and less listing overhead
• On managed tables with predictive optimization, Databricks does this for you

%sql
DESCRIBE DETAIL orders_delta

• Compare numFiles before and after

• Databricks recommends liquid clustering for all new tables
• It replaces partitioning and ZORDER. They can't be combined with it

%sql
CREATE OR REPLACE TABLE orders_clustered
CLUSTER BY (customer_id)
AS SELECT * FROM orders

%sql
OPTIMIZE orders_clustered

• Clustering happens when OPTIMIZE runs (or automatically with predictive optimization)
• Pick keys you filter or join on: up to 4, no complex types, columns need statistics
• Change later without rewriting history: ALTER TABLE orders_clustered CLUSTER BY (order_ts)
• Let Databricks choose: CLUSTER BY AUTO (needs predictive optimization)

%sql
CREATE OR REPLACE TABLE orders_cdf
TBLPROPERTIES (delta.enableChangeDataFeed = true)
AS SELECT * FROM orders WHERE order_id <= 5

%sql
UPDATE orders_cdf SET quantity = 9 WHERE order_id = 1

%sql
DELETE FROM orders_cdf WHERE order_id = 2

%sql
SELECT order_id, quantity, _change_type, _commit_version
FROM table_changes('orders_cdf', 0)
ORDER BY _commit_version, order_id

• Version 0: five insert rows. Version 1: a pre-image and post-image of order 1. Version 2: a delete of order 2
• Also: _commit_version and _commit_timestamp
• PySpark: .option("readChangeFeed", "true").option("startingVersion", 0).table(...)
• Use it to feed downstream tables with only what changed (Module 9, Module 21)

%sql
ANALYZE TABLE orders_delta COMPUTE STATISTICS FOR ALL COLUMNS

Maintenance checklist per table:

• Layout: liquid clustering on the columns you filter by
• Files: OPTIMIZE (or predictive optimization)
• Cleanup: VACUUM with at least 7 days retention
• Statistics: kept current so queries can skip files
• Retention: set logRetentionDuration and deletedFileRetentionDuration on purpose

Clean up the demo tables:

for t in ["orders_delta", "orders_clustered", "orders_cdf", "customers_delta", "big_orders"]:
    spark.sql(f"DROP TABLE IF EXISTS {t}")