Delta Lake Explained: Reliable Lakes with ACID

A data lake is cheap, scalable storage full of files. What it lacks by itself is the behaviour people expect from a database: a write that either happens completely or not at all, protection from bad data, and the ability to update rows. Delta adds those on top of ordinary Parquet files. Its documentation describes it as an open source project for building a lakehouse on data lakes, with ACID transactions, schema enforcement, time travel and merge, update and delete (Delta Lake docs: introduction). This guide shows each of those on a real table.

The idea: data files plus a transaction log

A Delta table is a folder of Parquet data files and a _delta_log folder. Every change is a commit: a JSON file in the log, named with the next version number padded to 20 digits, listing which data files were added and removed. The protocol calls these files the unit of atomicity: a table has a single serial history of atomic versions (Delta protocol). Readers build the current table from the log, so a half-finished write, whose commit file does not exist yet, is simply invisible. That is how a pile of files gains transactions.

A Delta table, step by step

The example uses the deltalake Python library, a Rust implementation of the format with Python bindings that needs no Spark or JVM (delta-rs docs), so it runs anywhere. The table holds daily trips and fares per pickup zone, built from January 2024 New York yellow taxi records (NYC TLC: trip record data). The same operations exist in Spark on Databricks and elsewhere.

import os
import shutil
import urllib.request
from pathlib import Path
import pandas as pd
from deltalake import DeltaTable, write_deltalake

source = "yellow_tripdata_2024-01.parquet"
if not Path(source).exists():
    urllib.request.urlretrieve(f"https://d37ci6vzurychx.cloudfront.net/trip-data/{source}", source)
raw = pd.read_parquet(source, columns=["tpep_pickup_datetime", "PULocationID", "fare_amount"])
raw = raw[raw["tpep_pickup_datetime"].between("2024-01-01", "2024-02-01", inclusive="left")]

def daily(trips):
    return (trips.assign(pickup_date=trips["tpep_pickup_datetime"].dt.strftime("%Y-%m-%d"))
                 .groupby(["pickup_date", "PULocationID"], as_index=False)
                 .agg(trips=("fare_amount", "size"), fare=("fare_amount", "sum")))

table = daily(raw)
path = "daily_zone_trips"
shutil.rmtree(path, ignore_errors=True)

write_deltalake(path, table[table["pickup_date"] <= "2024-01-15"], mode="overwrite")   # version 0
write_deltalake(path, table[table["pickup_date"] > "2024-01-15"], mode="append")       # version 1

try:                                                                                   # schema enforcement
    write_deltalake(path, table.head(3).assign(tip=0.0), mode="append")
except Exception as err:
    print("rejected:", type(err).__name__)

dt = DeltaTable(path)
dt.delete("PULocationID IN (264, 265)")                                                # version 2: unknown zones

fixed = daily(raw[(raw["tpep_pickup_datetime"] >= "2024-01-31") & (raw["fare_amount"] >= 0)])
metrics = (dt.merge(fixed, predicate="t.pickup_date = s.pickup_date AND t.PULocationID = s.PULocationID",
                    source_alias="s", target_alias="t")
             .when_matched_update_all()
             .when_not_matched_insert_all()
             .execute())                                                               # version 3: correct one day
print({k: metrics[k] for k in ("num_source_rows", "num_target_rows_updated", "num_target_rows_inserted")})

for entry in sorted(dt.history(), key=lambda h: h["version"]):
    print("version", entry["version"], entry["operation"])
print("log files:", sorted(f for f in os.listdir(f"{path}/_delta_log") if f.endswith(".json")))
for version in (1, 3):
    snapshot = DeltaTable(path, version=version).to_pandas()
    print(f"version {version}: {len(snapshot)} rows, {snapshot['trips'].sum()} trips, fares {snapshot['fare'].sum():,.2f}")

rejected: SchemaMismatchError
{'num_source_rows': 223, 'num_target_rows_updated': 221, 'num_target_rows_inserted': 2}
version 0 WRITE
version 1 WRITE
version 2 DELETE
version 3 MERGE
log files: ['00000000000000000000.json', '00000000000000000001.json', '00000000000000000002.json', '00000000000000000003.json']
version 1: 6949 rows, 2964606 trips, fares 53,881,922.66
version 3: 6889 rows, 2951717 trips, fares 53,554,194.24

What each step shows

Writes are versions. The overwrite and the append became versions 0 and 1, each one JSON file in the log. A reader that started during the append saw version 0 until version 1's commit file appeared, never half of it.

Schema enforcement. The third write tried to append rows with an extra column, tip, and was rejected with a SchemaMismatchError; no version was created, so the table was untouched. A plain folder of Parquet files would have accepted the new file, leaving readers to discover mixed schemas later. When a new column is intended, Delta supports evolving the schema explicitly, in deltalake with a schema mode of merge.

Delete. Zones 264 and 265 are the lookup table's unknown and outside-New-York zones, so they are removed. Parquet files are immutable, so a delete writes new files without the matching rows and records the swap as version 2.

Merge, and a lesson. The last step corrects 31 January by recomputing it without negative fares and merging: rows that match on date and zone are updated, new ones inserted. The metrics show 221 updated and 2 inserted. Those two inserts are the unknown zones deleted in version 2: the corrected source still contained them, and a merge applies its source exactly as given. The delete removed 62 rows and the merge put 2 of them back, which is why version 3 has 60 fewer rows than version 1. In production, the fix is one cleaning function shared by every path that writes the table, so a backfill cannot reintroduce what a rule removed.

Time travel. Reading version 1 and version 3 side by side shows the effect of the delete and the correction: 12,889 fewer trips and 327,728.42 less in fares. Being able to query an earlier version is invaluable for audits, for reproducing a report exactly as it was, and for recovering from a bad write by restoring the previous version.

Keeping Delta tables healthy

• Compact small files. Frequent small writes create many small files; periodic optimisation rewrites them into fewer large ones.
• Vacuum old files that no version still references, once they are older than your time-travel retention; after vacuuming, those old versions can no longer be read.
• Let checkpoints summarise the log: Delta periodically writes Parquet checkpoints so readers do not replay thousands of JSON commits.
• Partition or cluster large tables by columns commonly used in filters, such as date.
• Plan for concurrent writers. When two jobs change the same data at once, Delta detects the conflict and fails one commit instead of losing an update; while preparing this example, a delete run against a stale table handle failed exactly that way. Pipelines should reload the table and retry.

Where Delta fits among table formats

Delta is one of three open table formats; Apache Iceberg and Apache Hudi solve the same problems with different designs and ecosystems, compared in Iceberg vs Delta Lake vs Hudi. All three store data in Parquet, covered in Parquet vs ORC vs Avro. Delta is the native format of Databricks, introduced in the Databricks tutorial, and Spark reads and writes it at scale, as described in Apache Spark explained.

Data lake implementation is one of the projects, and Azure Databricks part of the Azure module, in the Data Engineering course.

Sources

• Introduction to Delta Lake (Delta Lake documentation) — Delta Lake
• Delta transaction log protocol (Delta Lake, GitHub) — Delta Lake (GitHub)
• delta-rs: the deltalake Python library (Delta Lake) — Delta Lake
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission