Delta Lake tables: the transaction log, time travel and RESTORE

Module 2 wrote the shop's orders as a folder of Parquet files. A folder is fine until something goes wrong while it is being rewritten, or until 1 order changes and you need to update it. This module turns the folder into a table, first with Delta Lake, the format Databricks uses for every table, and then looks at Iceberg.

Set up Delta Lake on your computer

On Databricks there is nothing to install: every table is a Delta table unless you ask for another format. On your computer, add the delta-spark package to the PySpark you installed in module 0. This module was run on PySpark 4.2.0 with delta-spark 4.4.1:

pip install delta-spark==4.4.1

Save this as delta_session.py. Every script in the module starts its session with it:

"""A local Spark session with Delta Lake. On Databricks, use the spark it gives you instead."""
from delta import configure_spark_with_delta_pip
from pyspark.sql import SparkSession


def delta_spark(app: str = "tables") -> SparkSession:
    builder = (
        SparkSession.builder.master("local[4]").appName(app)
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        .enableHiveSupport()
    )
    spark = configure_spark_with_delta_pip(builder).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")
    return spark

The 2 settings plug Delta into Spark's SQL and catalog. enableHiveSupport() keeps table names between runs, in a metastore_db folder, and the tables themselves go in spark-warehouse. The first run downloads Delta's Java files, so it needs the internet once.

What a failed rewrite leaves behind

Here are the 4 current orders from module 2, written twice: once as Parquet, once as a Delta table. Then each is rewritten by a job that fails halfway, on a bad row. Save this as orders_table.py:

import os
from pyspark.sql import functions as F
from delta_session import delta_spark

spark = delta_spark()
orders = spark.createDataFrame(
    [
        (1001, "C01", "shipped", 25.50, "2026-10-02 08:00:00"),
        (1002, "C02", "cancelled", 12.00, "2026-10-03 09:00:00"),
        (1004, "C01", "placed", 40.00, "2026-10-02 12:45:00"),
        (1005, "C09", "placed", 8.75, "2026-10-03 14:10:00"),
    ],
    "order_id BIGINT, customer_id STRING, status STRING, amount DOUBLE, updated_at STRING",
).select(
    "order_id", "customer_id", "status",
    F.col("amount").cast("decimal(10,2)").alias("amount"),
    F.col("updated_at").cast("timestamp").alias("updated_at"),
)

# A rewrite that fails halfway: 1 bad row stops the job.
broken = orders.withColumn(
    "amount", F.when(F.col("order_id") == 1005, F.raise_error(F.lit("bad row"))).otherwise(F.col("amount"))
)

orders.write.mode("overwrite").parquet("lake/orders_parquet")
try:
    broken.write.mode("overwrite").parquet("lake/orders_parquet")
except Exception as error:
    print("Parquet rewrite failed:", type(error).__name__)
print("Parquet files left:", os.listdir("lake/orders_parquet"))

orders.write.format("delta").mode("overwrite").saveAsTable("orders")
try:
    broken.write.format("delta").mode("overwrite").saveAsTable("orders")
except Exception as error:
    print("Delta rewrite failed:", type(error).__name__)
print("Delta rows:", spark.table("orders").count())

Parquet rewrite failed: SparkRuntimeException
Parquet files left: []
Delta rewrite failed: SparkRuntimeException
Delta rows: 4

The Parquet writer deleted the old files when the overwrite began, so the failed job left an empty folder, and every report that reads it fails until someone reruns the job. The Delta table still has its 4 rows.

The transaction log

The difference is a folder called _delta_log inside the table. Save this as log.py:

import json, os
from delta_session import delta_spark

spark = delta_spark()
location = spark.sql("DESCRIBE DETAIL orders").first().location.removeprefix("file:")
log = os.path.join(location, "_delta_log")
print(sorted(name for name in os.listdir(log) if name.endswith(".json")))
with open(os.path.join(log, "00000000000000000000.json")) as commit:
    for line in commit:
        action, body = next(iter(json.loads(line).items()))
        print(action, body.get("path", ""))

['00000000000000000000.json']
commitInfo
metaData
protocol
add part-00000-ea515a97-fdd1-4a96-ad79-b4352a6351d1-c000.snappy.parquet
add part-00001-b958d1c4-4e2c-4366-acb4-9972ca9a7415-c000.snappy.parquet
add part-00002-d2302cad-d0c1-4b6d-a26f-f5697f310047-c000.snappy.parquet
add part-00003-98a8b7b2-961d-4c50-8320-5d04c5fd3be5-c000.snappy.parquet

Each commit is 1 JSON file, numbered from 0. This one says what ran (commitInfo), the schema (metaData), the reader and writer versions it needs (protocol), and 1 add for each Parquet file. A write puts its data files in the folder first and commits last, by creating the next numbered file. The failed rewrite wrote files too, but never got to its commit, so it has no number and no reader looks at its files. That is what a transaction means here: the whole change or none of it.

The table's folder holds the 4 files version 0 lists, the 4 the failed job left, and the log. A reader never lists the folder; it reads the log.

History and time travel

Delta never edits a Parquet file. An update writes a new file with the changed rows and commits a remove for the old one. Save this as history.py:

from pyspark.sql import functions as F
from delta_session import delta_spark

spark = delta_spark()
spark.sql("UPDATE orders SET status = 'shipped', updated_at = '2026-10-04 08:30:00' WHERE order_id = 1004")
spark.sql("DELETE FROM orders WHERE status = 'cancelled'")

metrics = F.col("operationMetrics")
spark.sql("DESCRIBE HISTORY orders").select(
    "version", "operation",
    F.coalesce(metrics["numAddedFiles"], metrics["numFiles"]).alias("files_added"),
    metrics["numRemovedFiles"].alias("files_removed"),
).show()
spark.sql("SELECT order_id, status FROM orders VERSION AS OF 0 ORDER BY order_id").show()

+-------+--------------------+-----------+-------------+
|version|           operation|files_added|files_removed|
+-------+--------------------+-----------+-------------+
|      2|              DELETE|          0|            1|
|      1|              UPDATE|          1|            1|
|      0|CREATE OR REPLACE...|          4|            0|
+-------+--------------------+-----------+-------------+

+--------+---------+
|order_id|   status|
+--------+---------+
|    1001|  shipped|
|    1002|cancelled|
|    1004|   placed|
|    1005|   placed|
+--------+---------+

Each order sits in a file of its own here, so the update of 1004 replaced 1 file and the delete removed 1. Version 0 still reads as it was, because its files are still on disk: VERSION AS OF replays the log only up to that commit. TIMESTAMP AS OF with a date and time picks the version that was current then.

Try itA Delta table's log, version by version

Read version

0123

VACUUM RETAIN 0 HOURS

Not runRun

_delta_log

1. 000CREATE OR REPLACE TABLE AS SELECT+ 1001 shipped + 1002 cancelled + 1004 placed + 1005 placed
2. 001UPDATE 1004 to shipped+ 1004 shipped − 1004 placed
3. 002DELETE the cancelled orders− 1002 cancelled
4. 003RESTORE TO VERSION AS OF 0+ 1002 cancelled + 1004 placed − 1004 shipped

Parquet files in the folder

• part-00000-e59f…1001 shippedRead
• part-00001-183b…1002 cancelledRead
• part-00002-9bbc…1004 placedRead
• part-00003-1240…1005 placedRead
• part-00000-e97b…1004 shippedNot in v3
• 4 filesthe failed rewriteNot in v3

SELECT * FROM orders VERSION AS OF 3

1001 shipped · 1002 cancelled · 1004 placed · 1005 placed

Put a version back with RESTORE

To undo the update and the delete, restore version 0. Save this as restore.py:

from delta_session import delta_spark

spark = delta_spark()
spark.sql("RESTORE TABLE orders TO VERSION AS OF 0").select(
    "num_removed_files", "num_restored_files"
).show()

+-----------------+------------------+
|num_removed_files|num_restored_files|
+-----------------+------------------+
|                1|                 2|
+-----------------+------------------+

RESTORE is a new commit, version 3, that removes the shipped 1004 file and adds back the 2 files version 0 had. The update and the delete stay in the history, so you can undo the restore too.

On Databricks, the same SQL works on a table in your catalog and schema, and Databricks manages where its files live, so you read the log through DESCRIBE HISTORY rather than its JSON files. History is kept for 30 days by default and the files of old versions for 7, until VACUUM removes them, which part 4 covers. Databricks says not to use history as a backup.

Part 2 updates the table from a batch of changes with MERGE.