Delta Lake & the Lakehouse

In "Storage Layers & File Formats" you met Parquet files and the small-files problem. A folder of Parquet files is cheap. But it is only a folder of files.

Plain files have no transactions. A transaction is a change that happens fully or not at all.

A job that fails halfway leaves half-written files. Two jobs that write at the same time can break each other.

Delta Lake fixes this. It keeps a small log next to the Parquet files.

The log says which files make up the table. The log and its rules are called a table format.

A lakehouse joins three parts.

1. Open files, such as Parquet, on cheap storage.
2. A table format, such as Delta Lake, that adds transactions.
3. Engines, such as Spark, that read and write the table.

In "The Warehouse Platform: Storage, Compute & Loading" you saw time travel in Snowflake. Delta Lake gives you time travel on files that you own.

This chapter uses open-source Delta Lake 3.x. You turn it on once, when you build the Spark session. You also need the delta-spark package.

from pyspark.sql import SparkSession

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

The data is the shop again. The table orders starts with six rows.

You will change this one Delta table, one version at a time. Here is the whole story.

A Delta table is a folder. It holds Parquet data files and a folder named _delta_log.

Every change adds one new file to _delta_log. That file is a commit.

A commit is a small JSON file. It lists actions, such as "add this data file" and "remove that data file". The file name is the version number, with 20 digits.

The data files are called A, B and A2 here. Real names are long and random.

Look at versions 0, 1 and 2. The code makes three changes.

orders_df.write.format("delta").saveAsTable("orders")           # version 0

new_orders.write.format("delta").mode("append").saveAsTable("orders")   # version 1

spark.sql("""UPDATE orders
             SET status = 'paid', amount = 22.00
             WHERE order_id = 105""")                                  # version 2

To read version 2, Delta reads commits 0, 1 and 2 in order. It keeps every file that was added and not removed. That gives files B and A2: 2 + 6 = 8 rows.

A write has two steps. First, Spark writes the Parquet files. Last, Delta creates the commit file.

Until the commit file exists, readers do not see the new files.

That is why a failed job leaves no half-written table. The commit never appears. This is one of the four ACID promises.

• Atomic: a write happens fully or not at all.
• Consistent: the table always follows its rules.
• Isolated: a reader sees one whole version, never a half-finished write.
• Durable: after a commit, the data stays.

Replaying thousands of commits is slow. So every 10 commits, Delta writes a checkpoint. It is a Parquet file that lists all live files at that version.

The setting is delta.checkpointInterval, and its default is 10. A reader starts at the newest checkpoint and reads only the later commits.

Our story stops at version 9, so it has no checkpoint yet. Version 10 would add the file 00000000000000000010.checkpoint.parquet.

The log keeps every version, and the old data files stay in the folder. So you can read the table as it was. This is time travel.

First, find the versions. DESCRIBE HISTORY lists every commit, newest first. The timestamps are made up.

DESCRIBE HISTORY orders

Now read old versions. Compare the paid revenue at version 1 with the paid revenue now. Order 105 was paid at version 2, so the number differs.

SELECT SUM(amount) FROM orders VERSION AS OF 1 WHERE status = 'paid';

SELECT SUM(amount) FROM orders TIMESTAMP AS OF '2026-01-22 11:00:00'
WHERE status = 'paid';

SELECT SUM(amount) FROM orders WHERE status = 'paid';   -- version 2

Time travel also saves you from mistakes. At version 3, someone deletes customer 3 by mistake. Then RESTORE puts the table back.

DELETE FROM orders WHERE customer_id = 3;        -- version 3

RESTORE TABLE orders TO VERSION AS OF 2;         -- version 4

Delta uses optimistic concurrency. Each job reads the newest version, does its work, and then tries to create the next commit file. Only one job can create a file with a given name.

The job that loses checks one thing. Did the winning commit touch anything my job read?

If not, the loser tries again as the next version. If yes, it stops with an error.

These two cases are side stories. They do not change the table in our story. Both start when orders is at version 4.

# Case 1: two jobs only append
job_a: append order 230
job_b: append order 231

# Case 2: one job deletes, one updates
job_a: DELETE FROM orders WHERE status = 'refunded'
job_b: UPDATE orders SET status = 'refunded' WHERE order_id = 106   # commits first

Delta has several conflict errors. Three are common.

To avoid conflicts, partition the table. A partition is a folder for the rows with one value of a column. Then put the partition column in the condition of your DELETE, UPDATE or MERGE.

Delta then knows the jobs touch different partitions. A simple retry also works.

On Amazon S3, writes from more than one cluster need extra setup. Check the Delta docs for your storage.

A schema is the list of columns and their types. Delta enforces it.

A write with an extra column, or a different type, is rejected. This stops bad data from entering quietly.

Look at one new order, 222. It has an extra column, coupon. The table does not have it, so the append fails.

from datetime import date
from decimal import Decimal

new = spark.createDataFrame(
    [(222, 1, "paid", Decimal("18.00"), date(2026, 1, 23), "WINTER")],
    "order_id INT, customer_id INT, status STRING, "
    "amount DECIMAL(10,2), order_date DATE, coupon STRING")

new.write.format("delta").mode("append").saveAsTable("orders")

Sometimes you want the new column. Then you say so, with the option mergeSchema. This is schema evolution.

(new.write.format("delta").mode("append")
    .option("mergeSchema", "true")
    .saveAsTable("orders"))                       # version 5

spark.sql("SELECT order_id, coupon FROM orders ORDER BY order_id")

In "Loading & Write Patterns" you met the upsert: update a row if it exists, insert it if not. In "Change Data Capture (CDC)" you saw change rows with an operation flag. MERGE INTO applies such rows to a Delta table in one commit.

Look at three change rows. The flag op says what to do: u is update, i is insert, d is delete.

MERGE INTO orders t
USING changes s
  ON t.order_id = s.order_id
WHEN MATCHED AND s.op = 'd' THEN DELETE
WHEN MATCHED THEN
  UPDATE SET t.status = s.status, t.amount = s.amount
WHEN NOT MATCHED AND s.op = 'i' THEN
  INSERT (order_id, customer_id, status, amount, order_date)
  VALUES (s.order_id, s.customer_id, s.status, s.amount, s.order_date)

Delta rewrites only the files that hold changed rows. Version 6 removes file A2 (order 104) and file B (order 221).

It adds A4, B2 and D. File C is untouched.

A third clause is WHEN NOT MATCHED BY SOURCE. It acts on target rows that have no source row. It needs Delta 2.3 or later.

MERGE INTO orders t
USING full_snapshot s
  ON t.order_id = s.order_id
WHEN NOT MATCHED BY SOURCE THEN DELETE

Each small commit makes small files, which you saw hurt reads in "Storage Layers & File Formats". OPTIMIZE joins them into bigger files.

Now look at the files. Version 6 left four small files.

OPTIMIZE orders          -- version 7

OPTIMIZE does not delete the old files. They stay for time travel. They also cost storage until VACUUM removes them.

Sometimes a job downstream needs only the rows that changed. In "Change Data Capture (CDC)" the source was a database log. Delta has its own: the change data feed.

You turn it on for a table. From then on, Delta also records every changed row.

Look at versions 8 and 9. Order 223 changes from paid to refunded.

spark.sql("ALTER TABLE orders SET TBLPROPERTIES (delta.enableChangeDataFeed = true)")   # version 8
spark.sql("UPDATE orders SET status = 'refunded' WHERE order_id = 223")                 # version 9

changes = (spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 9)
    .table("orders")
    .select("order_id", "status", "amount", "_change_type", "_commit_version"))

Old data files pile up. Every update, MERGE and OPTIMIZE leaves files that the newest version no longer uses. They cost storage.

VACUUM deletes data files that the newest version does not use. They must also be older than the retention time.

The default retention is 7 days, which is 168 hours.

Say it is 2026-02-02, 10:00. Nine files are unused, and only file F is live. Look at every data file in the story.

VACUUM orders DRY RUN;    -- lists the files, deletes nothing

VACUUM orders;            -- 168 hours by default

Two settings control how far back you can go. If you run VACUUM often, the shorter one wins.

Delta refuses a VACUUM with less than 7 days of retention. A running job may still read those files. You can switch the check off with spark.databricks.delta.retentionDurationCheck.enabled, but do it only with care.

Longer retention means more storage cost and more time travel. Match it to how far back your team must audit or restore.

Delta stores the smallest and largest value of the first 32 columns, for each file, in the log. These are file statistics. A query with a filter can skip a file when the value cannot be inside it.

Statistics help only when similar values sit together. OPTIMIZE with ZORDER BY sorts the data by a column, so each file holds a small range.

This example is made up and bigger than our table. Three files each hold customers 1 to 4. The query asks for customer 1, Asha.

OPTIMIZE orders ZORDER BY (customer_id);

SELECT * FROM orders WHERE customer_id = 1;

Z-ORDER has a cost. It rewrites the files again, every time you run it. You also must name the columns again in every run.

Liquid clustering is a newer way. It came as a preview in Delta Lake 3.1 and is ready for use from 3.2. You name the columns when you create the table.

CREATE TABLE orders_clustered (
  order_id INT, customer_id INT, status STRING,
  amount DECIMAL(10,2), order_date DATE
) USING DELTA
CLUSTER BY (customer_id);

OPTIMIZE orders_clustered;     -- clusters the new data

An update or delete normally rewrites a whole data file, even for one row. On big files this is slow and costly.

A deletion vector is a small side file. It marks rows as deleted, so Delta does not rewrite the data file. Readers skip the marked rows.

This is a side story, from the table at version 5. You delete order 221. Its file B holds two rows: 220 and 221.

ALTER TABLE orders
SET TBLPROPERTIES (delta.enableDeletionVectors = true);

DELETE FROM orders WHERE order_id = 221;

REORG TABLE orders APPLY (PURGE);   -- later: rewrite the files

Delta Lake is one table format. Apache Iceberg and Apache Hudi are two others. All three are open, and all three add table rules to files on cheap storage.

They differ in how they keep the list of files. Delta uses the JSON log you saw.

Iceberg keeps the list in its own metadata files. Hudi keeps a timeline of actions.

Pick by the engines and tools your team already uses. Check that each tool you need can read and write the format.

Two more points. Auto Loader loads new files into a table as they arrive. It is a Databricks feature, not part of open-source Delta.

Delta is also a stream source and a stream sink. In "Structured Streaming Fundamentals" you used foreachBatch with MERGE as a sink that is safe to run twice. A Delta table is the table you MERGE into.

To control how Spark writes files, see "Reading & Writing Data: Sources & File Formats". It covers save modes, partitionBy and maxRecordsPerFile.

Six points to keep.

• A Delta table is Parquet files plus a log. Each commit adds or removes files.
• A write appears only when its commit file is created. A writer fails only if another commit changed data that it read.
• Old versions stay for time travel and RESTORE, until VACUUM deletes the files.
• Schema enforcement rejects bad writes. mergeSchema adds columns when you ask.
• MERGE applies inserts, updates and deletes in one commit. OPTIMIZE joins small files.
• With regular VACUUM and default settings, time travel reaches back about 7 days.

Answer these without looking back.

1. Which files does version 2 of orders use, and how many rows do they hold?
2. Why does a failed write leave no half-written table?
3. Job A deletes refunded orders. Job B updates order 106 and commits first. What happens to job A?
4. Why does the append of order 222 fail, and how do you allow it?
5. After VACUUM on 2026-02-02, can you read version 6? Why?