Delta Lake: Reliable Lakehouse Storage with ACID Transactions

Data lakes built on object storage are cheap and scalable, but plain file formats give you no guarantees: a half-finished write can leave readers with corrupt data, and concurrent jobs can clobber each other. Delta Lake fixes this by adding a transaction log on top of Parquet, bringing ACID transactions, schema enforcement, and time travel to the lake. This tutorial walks through Delta Lake from first principles using two practical paths: the lightweight Python deltalake library (no Spark required) and Apache Spark with delta-spark.

What Delta Lake Is

Delta Lake is an open table format. A Delta table is not a single file but a directory that contains:

• One or more Parquet data files holding the actual rows.
• A transaction log directory named deltalog, containing ordered JSON commit files (00000000000000000000.json, 00000000000000000001.json, and so on) plus periodic Parquet checkpoints.

Every change to the table is recorded as an atomic commit in the log. Readers consult the log to learn exactly which Parquet files make up the current version of the table. Because the log is the source of truth, a reader never sees a partially written file, and two writers can detect that they conflicted.

This is the core of the lakehouse idea: keep the low cost and openness of a data lake (Parquet on S3, ADLS, GCS, or local disk), but add the reliability and transactional semantics you would expect from a data warehouse.

Why Not Just Parquet?

Plain Parquet is excellent columnar storage, but it has no notion of a table version or a commit. If a job writes ten Parquet files and crashes after five, a reader scanning the directory will pick up an inconsistent mix. There is no atomic "append", no safe concurrent update, no way to roll back, and schema changes are entirely manual. Delta Lake keeps Parquet as the storage layer and layers transactional metadata on top.

The Transaction Log in Detail

A typical Delta directory looks like this:

orders/
  deltalog/
    00000000000000000000.json
    00000000000000000001.json
    00000000000000000002.json
    lastcheckpoint
  part-00000-....snappy.parquet
  part-00001-....snappy.parquet

Each JSON commit describes actions: files added (add), files removed (remove), metadata changes, and protocol changes. Version 0 is the table creation; each subsequent commit increments the version by one. This monotonic version number is what makes time travel possible, and the remove actions are what let VACUUM later reclaim disk space.

Path 1: The Python deltalake Library (No Spark)

The deltalake package (the delta-rs project) is a native Rust implementation with Python bindings. It reads and writes Delta tables without a JVM or Spark cluster, which makes it ideal for single-node ETL, notebooks, and embedding Delta into existing pandas or PyArrow pipelines.

Installation

pip install "deltalake>=0.18" pandas pyarrow

Creating and Appending to a Table

We will model an orders table throughout this tutorial.

import pandas as pd
from deltalake import writedeltalake, DeltaTable


tablepath = "data/orders"

orders = pd.DataFrame(
    {
        "orderid": [1, 2, 3],

        "customer": ["alice", "bob", "carol"],
        "amount": [120.0, 80.5, 45.0],
        "status": ["paid", "paid", "pending"],
        "region": ["west", "east", "west"],
    }
)

First write creates the table at version 0.
writedeltalake(tablepath, orders, mode="overwrite")

neworders = pd.DataFrame(
    {
        "orderid": [4, 5],

        "customer": ["dave", "erin"],
        "amount": [200.0, 15.0],
        "status": ["paid", "paid"],
        "region": ["east", "west"],
    }
)

writedeltalake(tablepath, neworders, mode="append")

Reading the Table

dt = DeltaTable(tablepath)

As a pandas DataFrame
df = dt.topandas()
print(df)

As a PyArrow table (useful for streaming into other engines)
arrowtable = dt.topyarrowtable()


Inspect the current version
print("current version:", dt.version())

paid = dt.topandas(
    columns=["orderid", "customer", "amount"],

    filters=[("status", "=", "paid")],
)

# This would raise because discount is a new column.
bad = pd.DataFrame({"orderid": [6], "discount": [0.1]})
writedeltalake(tablepath, bad, mode="append")  # -> schema mismatch error

When a schema change is intentional, opt in explicitly with schemamode="merge":

extended = pd.DataFrame(
    {
        "orderid": [6],
        "customer": ["frank"],
        "amount": [99.0],
        "status": ["paid"],
        "region": ["east"],
        "discount": [0.10],  # brand-new column
    }
)

writedeltalake(tablepath, extended, mode="append", schemamode="merge")

Existing rows get null for the new discount column. Use schemamode="overwrite" together with mode="overwrite" only when you deliberately want to replace the schema.

Time Travel with deltalake

Because every commit is versioned, you can load any historical state of the table.

# Load a specific version
v0 = DeltaTable(tablepath, version=0).topandas()

Or load as of a timestamp
dt = DeltaTable(tablepath)

dt.loadasversion(1)

snapshot = dt.topandas()

Inspect the full history
for commit in DeltaTable(tablepath).history():

    print(commit.get("version"), commit.get("operation"), commit.get("timestamp"))

dt = DeltaTable(tablepath)

DELETE rows matching a predicate
dt.delete("status = 'pending'")

UPDATE rows
dt.update(
    predicate="region = 'west'",
    updates={"status": "'shipped'"},
)

Upserts use the merge API. The pattern is: match on a key, update when matched, insert when not matched.

import pyarrow as pa

updates = pa.table(
    {
        "orderid": [4, 7],          # 4 exists -> update; 7 is new -> insert

        "customer": ["dave", "gina"],
        "amount": [222.0, 60.0],
        "status": ["refunded", "paid"],
        "region": ["east", "west"],
    }
)

(
    DeltaTable(tablepath)
    .merge(
        source=updates,
        predicate="target.orderid = source.orderid",
        sourcealias="source",

        targetalias="target",
    )
    .whenmatchedupdateall()

    .whennotmatchedinsertall()

    .execute()
)

dt = DeltaTable(tablepath)

Compact small files into larger ones
dt.optimize.compact()

Remove data files no longer referenced by the log.
retentionhours guards against deleting files still needed for time travel.

dt.vacuum(retentionhours=168, dryrun=False, enforceretentionduration=True)

pip install pyspark==3.5.1 delta-spark==3.2.0

from pyspark.sql import SparkSession
from delta import configuresparkwithdeltapip


builder = (
    SparkSession.builder.appName("delta-orders")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.sparkcatalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    )
)

spark = configuresparkwithdeltapip(builder).getOrCreate()

Creating and Appending with Spark

data = [
    (1, "alice", 120.0, "paid", "west"),
    (2, "bob", 80.5, "paid", "east"),
    (3, "carol", 45.0, "pending", "west"),
]
columns = ["orderid", "customer", "amount", "status", "region"]


df = spark.createDataFrame(data, columns)

Write a partitioned Delta table
(
    df.write.format("delta")
    .mode("overwrite")
    .partitionBy("region")
    .save("/data/ordersspark")
)

Append more rows
more = spark.createDataFrame([(4, "dave", 200.0, "paid", "east")], columns)
more.write.format("delta").mode("append").save("/data/ordersspark")

CREATE TABLE orders
USING DELTA
LOCATION '/data/ordersspark';

Schema Evolution with mergeSchema

from pyspark.sql.functions import lit

evolved = (
    spark.createDataFrame([(5, "erin", 30.0, "paid", "west")], columns)
    .withColumn("discount", lit(0.05))
)

(
    evolved.write.format("delta")
    .mode("append")
    .option("mergeSchema", "true")
    .save("/data/ordersspark")

)

# By version
v0 = spark.read.format("delta").option("versionAsOf", 0).load("/data/ordersspark")

By timestamp
ts = (
    spark.read.format("delta")
    .option("timestampAsOf", "2026-06-01 00:00:00")
    .load("/data/ordersspark")

)

SELECT  FROM orders VERSION AS OF 0;
SELECT  FROM orders TIMESTAMP AS OF '2026-06-01 00:00:00';
DESCRIBE HISTORY orders;

MERGE INTO orders AS target
USING updates AS source
ON target.orderid = source.orderid

WHEN MATCHED AND source.status = 'cancelled' THEN
  DELETE
WHEN MATCHED THEN
  UPDATE SET
    target.amount = source.amount,
    target.status = source.status
WHEN NOT MATCHED THEN
  INSERT (orderid, customer, amount, status, region)
  VALUES (source.orderid, source.customer, source.amount, source.status, source.region);

from delta.tables import DeltaTable as SparkDeltaTable

target = SparkDeltaTable.forPath(spark, "/data/ordersspark")
source = spark.createDataFrame(
    [(4, "dave", 222.0, "refunded", "east")], columns
)

(
    target.alias("t")
    .merge(source.alias("s"), "t.orderid = s.orderid")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute()
)

DELETE and UPDATE in Spark

DELETE FROM orders WHERE status = 'pending';

UPDATE orders SET status = 'shipped' WHERE region = 'west';

OPTIMIZE, Z-Ordering, and VACUUM

-- Compact small files
OPTIMIZE orders;

-- Co-locate related data to speed up filtered reads
OPTIMIZE orders ZORDER BY (customer);

-- Reclaim space from files older than the retention window
VACUUM orders RETAIN 168 HOURS;

Z-ordering reorganizes data within files so that rows with similar values in the Z-order columns sit together. When you later filter on those columns, Delta can skip more files via data-skipping statistics. It is most useful for high-cardinality columns you filter on often, such as customer or orderid.

from deltalake import DeltaTable

df = DeltaTable("/data/ordersspark").topandas(filters=[("region", "=", "west")])

INSTALL delta;
LOAD delta;

SELECT region, count(*) AS n, sum(amount) AS total
FROM deltascan('/data/ordersspark')

GROUP BY region
ORDER BY total DESC;

• Partition by low-cardinality columns you filter on frequently (for example region or an event date), and avoid over-partitioning, which creates many tiny files.
• Use MERGE INTO for idempotent ingestion so that reprocessing the same batch does not create duplicates.
• Run OPTIMIZE regularly on tables that receive many small appends, and add ZORDER BY on the columns you filter most.
• Set a sensible VACUUM retention (the default is 7 days). Do not lower it below the longest time-travel window you rely on, or you will break historical reads.
• Opt into schema evolution explicitly with mergeSchema / schemamode="merge" rather than turning it on globally, so accidental schema drift is still caught.
• Keep the writer versions aligned: with delta-spark, let configuresparkwithdeltapip manage the package version; with deltalake, pin a minimum version in your requirements.
• Prefer delta-rs for single-node jobs and Spark for distributed workloads; both read and write the same on-disk format, so you can mix them.

Conclusion and Key Takeaways

Delta Lake turns a directory of Parquet files into a reliable, transactional table by adding a versioned deltalog. The key points to remember:

• A Delta table is Parquet data plus a transaction log; the log is the source of truth and the basis for ACID guarantees.
• You have two practical entry points: the deltalake Python library for Spark-free, single-node work, and delta-spark for distributed processing and SQL.
• Schema enforcement protects you by default; opt into evolution deliberately.
• Time travel lets you read any past version by number or timestamp, which is invaluable for debugging and reproducibility.
• MERGE, UPDATE, and DELETE are first-class, and OPTIMIZE/Z-ordering plus VACUUM keep tables fast and tidy.
• Compared to plain Parquet, Delta adds reliability with little overhead; compared to Iceberg and Hudi, it is a strong choice for Spark- and Python-centric teams.

With these building blocks you can run a dependable lakehouse on commodity object storage, and read the same tables from Spark, pandas, or DuckDB as your needs change.