Sail examples

Appearance

advancedView Python source

Manage a Delta Lake table through its life

A table changes as data arrives, a new column appears, a month gets rebuilt, labels are fixed, rows are removed, and late corrections come in. This example takes a Delta Lake table through each of those changes and reads all eight versions it creates. The table holds daily trips per pickup borough, built from real NYC green taxi records for January to March 2024.

Run it

Run all commands below from the repository root (sail-examples/):

uv run pipelines/delta-table-lifecycle/main.py

The first run downloads three months of trip records and the zone lookup (4 MB) into pipelines/delta-table-lifecycle/.data/. Sail reads and writes Delta Lake without extra packages.

The table ends at version 7. Reading version 3 shows March 31 before the corrections: Manhattan with 740 trips, and the borough labels N/A and Unknown that later versions relabeled or removed.

Output

The newest capture appears first. Older output stays here when a result changes.

Current output · Sail 0.7.3

Captured on Sail 0.7.3. Latest capture: 2026-11-05, Python 3.12.14.

versions
+-------+----+
|version|rows|
+-------+----+
|0      |0   |
|1      |182 |
|2      |353 |
|3      |531 |
|4      |531 |
|5      |531 |
|6      |440 |
|7      |441 |
+-------+----+

by_month
+-----+-----+--------------+
|month|trips|rows_with_tips|
+-----+-----+--------------+
|1    |56437|151           |
|2    |53448|0             |
|3    |57575|148           |
+-----+-----+--------------+

march_31
+---------+-------------+-----+--------+
|version  |borough      |trips|fares   |
+---------+-------------+-----+--------+
|current  |Bronx        |17   |347.64  |
|current  |Brooklyn     |267  |6939.9  |
|current  |Manhattan    |1000 |15000.0 |
|current  |Outside NYC  |1    |69.0    |
|current  |Queens       |446  |8526.5  |
|current  |Staten Island|1    |20.0    |
|version 3|Bronx        |17   |347.64  |
|version 3|Brooklyn     |267  |6939.9  |
|version 3|Manhattan    |740  |10492.56|
|version 3|N/A          |1    |69.0    |
|version 3|Queens       |446  |8526.5  |
|version 3|Unknown      |4    |129.8   |
+---------+-------------+-----+--------+

boroughs
+-------------+
|borough      |
+-------------+
|Bronx        |
|Brooklyn     |
|EWR          |
|Manhattan    |
|Outside NYC  |
|Queens       |
|Staten Island|
+-------------+

Python code

main.py165 lines
View main.py on GitHub 
# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "pysail==0.7.3",
#   "pyspark-client==4.1.3",
#   "pyarrow==26.0.0",
#   "pandas==3.0.6",
# ]
# ///
"""The life of a Delta Lake table: append, evolve the schema, update, delete, merge, rebuild, time travel.

The table holds daily taxi trips per pickup borough, built from real NYC green taxi records for
January to March 2024. The example reads each of the eight table versions it creates.
"""

import atexit
import hashlib
import os
import shutil
import tempfile
import urllib.request
from pathlib import Path

from pyspark.sql import DataFrame, SparkSession

DATA = Path(__file__).resolve().parent / ".data"
SOURCE = "https://d37ci6vzurychx.cloudfront.net"
FILES = {
    # name: (url, SHA-256 of the file this example was written against)
    "green_tripdata_2024-01.parquet": "1512a953bed564ac68dcb622827424dfb4f68ad93c2a4572e50ef6ab0b09c1ad",
    "green_tripdata_2024-02.parquet": "cb7eb405ee7e795c9fd6b8ad1b291f46f4e0a1cd60a57067b0aaca2420e8bc5d",
    "green_tripdata_2024-03.parquet": "43dedd66dd556acfc38da679a174b6adc32e01dd6dccad14bdfdf1521a14abed",
    "taxi_zone_lookup.csv": "1a99e105092230f8620f301edcca7f80d3080642ff404d28ed957d3fa222c8ed",
}


def temp_dir(prefix: str) -> str:
    """A temporary directory that is removed when the process exits."""
    path = tempfile.mkdtemp(prefix=prefix)
    atexit.register(shutil.rmtree, path, ignore_errors=True)
    return path


def download(name: str) -> Path:
    """Reuse or download a file only when its SHA-256 matches the pinned input."""
    expected = FILES[name]
    folder = "misc" if name.endswith(".csv") else "trip-data"
    url = f"{SOURCE}/{folder}/{name}"
    path = DATA / name
    if not path.exists():
        path.parent.mkdir(parents=True, exist_ok=True)
        temporary = None
        try:
            with (
                urllib.request.urlopen(url, timeout=120) as response,
                tempfile.NamedTemporaryFile(dir=path.parent, prefix=f".{path.name}.", delete=False) as out,
            ):
                temporary = Path(out.name)
                while chunk := response.read(1024 * 1024):
                    out.write(chunk)
            with temporary.open("rb") as downloaded:
                if hashlib.file_digest(downloaded, "sha256").hexdigest() != expected:
                    raise ValueError(f"SHA-256 mismatch: {name}")
            os.replace(temporary, path)
        finally:
            if temporary is not None:
                temporary.unlink(missing_ok=True)
    with path.open("rb") as cached:
        if hashlib.file_digest(cached, "sha256").hexdigest() != expected:
            raise ValueError(f"SHA-256 mismatch: {name}")
    return path


def daily_by_borough(spark: SparkSession, month: int, with_tips: bool) -> DataFrame:
    """One month of trips summarized per day and pickup borough."""
    trips = spark.read.parquet(str(download(f"green_tripdata_2024-{month:02d}.parquet")))
    trips.createOrReplaceTempView("month_trips")
    tips = ", round(sum(t.tip_amount), 2) AS tips" if with_tips else ""
    return spark.sql(f"""
        SELECT {month} AS month, date(t.lpep_pickup_datetime) AS day, z.Borough AS borough,
               count(*) AS trips, round(sum(t.fare_amount), 2) AS fares{tips}
        FROM month_trips t JOIN zones z ON t.PULocationID = z.LocationID
        WHERE month(t.lpep_pickup_datetime) = {month} AND year(t.lpep_pickup_datetime) = 2024
        GROUP BY 2, 3
    """)


def run(spark: SparkSession) -> dict[str, DataFrame]:
    spark.read.schema("LocationID int, Borough string, Zone string, service_zone string").csv(
        str(download("taxi_zone_lookup.csv")), header=True
    ).createOrReplaceTempView("zones")
    path = temp_dir("delta-") + "/daily"

    # Version 0: declare the table: its columns, their types, and the partition column.
    spark.sql(f"""
        CREATE TABLE daily_trips (month INT, day DATE, borough STRING, trips BIGINT, fares DOUBLE)
        USING delta PARTITIONED BY (month) LOCATION '{path}'
    """)
    # Versions 1 and 2: append January, then February.
    for month in (1, 2):
        daily_by_borough(spark, month, with_tips=False).write.format("delta").mode("append").save(path)
    # Version 3: March arrives with a new column. mergeSchema adds it; older rows read it as NULL.
    march = daily_by_borough(spark, 3, with_tips=True)
    march.write.format("delta").mode("append").option("mergeSchema", "true").save(path)
    # Version 4: rebuild January with the tips column, replacing only that month's rows.
    january = daily_by_borough(spark, 1, with_tips=True)
    january.write.format("delta").mode("overwrite").option("replaceWhere", "month = 1").save(path)

    # Version 5: relabel a value in place.
    spark.sql("UPDATE daily_trips SET borough = 'Outside NYC' WHERE borough = 'N/A'")
    # Version 6: remove rows that should not be reported.
    spark.sql("DELETE FROM daily_trips WHERE borough = 'Unknown'")
    # Version 7: late corrections for one day: update the matching row, insert the missing one.
    corrections = spark.createDataFrame(
        [
            (3, "2024-03-31", "Manhattan", 1000, 15000.0, 1500.0),
            (3, "2024-03-31", "Staten Island", 1, 20.0, 0.0),
        ],
        "month int, day string, borough string, trips long, fares double, tips double",
    ).selectExpr("month", "CAST(day AS DATE) AS day", "borough", "trips", "fares", "tips")
    corrections.createOrReplaceTempView("corrections")
    spark.sql("""
        MERGE INTO daily_trips t USING corrections c
        ON t.day = c.day AND t.borough = c.borough
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)
    # Time travel: every version is still readable.
    versions = spark.createDataFrame(
        [(v, spark.read.format("delta").option("versionAsOf", v).load(path).count()) for v in range(8)],
        "version int, rows long",
    )
    current = spark.read.format("delta").load(path)
    summary = (
        current.groupBy("month")
        .agg({"trips": "sum", "tips": "count"})
        .withColumnsRenamed({"sum(trips)": "trips", "count(tips)": "rows_with_tips"})
    )
    march_31 = spark.sql(f"""
        SELECT 'version 3' AS version, borough, trips, fares FROM delta.`{path}` VERSION AS OF 3
        WHERE day = DATE'2024-03-31'
        UNION ALL
        SELECT 'current', borough, trips, fares FROM daily_trips WHERE day = DATE'2024-03-31'
    """)
    return {
        "versions": versions.orderBy("version"),
        "by_month": summary.orderBy("month"),
        "march_31": march_31.orderBy("version", "borough"),
        "boroughs": current.select("borough").distinct().orderBy("borough"),
    }


if __name__ == "__main__":
    from pysail.spark import SparkConnectServer

    server = SparkConnectServer()
    server.start()
    _, port = server.listening_address
    spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
    results = run(spark)
    for name, frame in results.items():
        print(name)
        frame.show(truncate=False)
    spark.stop()
    server.stop()

How it works

Version Change How
0 Declare the table CREATE TABLE ... (columns) USING delta PARTITIONED BY (month) LOCATION ...
1, 2 Append January, then February df.write.format("delta").mode("append")
3 Append March with a new tips column .option("mergeSchema", "true"): older rows read tips as NULL
4 Rebuild January, now with tips .mode("overwrite").option("replaceWhere", "month = 1") replaces only January
5 Relabel N/A as Outside NYC UPDATE daily_trips SET ... WHERE ...
6 Remove the Unknown borough DELETE FROM daily_trips WHERE ...
7 Late corrections for March 31 MERGE INTO ... WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *
  • Time travel. option("versionAsOf", v) reads version v, and SQL can do the same with VERSION AS OF. The versions output counts the rows in each version, and march_31 puts one day from version 3 next to the current table.
  • Declare first. Creating the table with its columns, types and partitioning before writing makes the schema a deliberate decision. Columns declared this way are nullable.

On Apache Spark

The same code runs on Apache Spark with the delta-spark package, its session extension (io.delta.sql.DeltaSparkSessionExtension) and its catalog (org.apache.spark.sql.delta.catalog.DeltaCatalog as spark_catalog). The checks run it on Spark 4.2 with delta-spark 4.4.1 and compare every result with Sail's.

On Sail

  • On Sail, reads of some older versions after a Delta checkpoint can fail. This example checks versions 0 through 7 before the first checkpoint. Verify reads of the specific versions you need in longer table histories.
  • To add a column with an append and mergeSchema, do it the way this example does: a table created with CREATE TABLE ... (columns) USING delta, then an append whose new column is computed (here sum(tip_amount)). ALTER TABLE ... ADD COLUMNS also works.
  • Sail does not support RESTORE. Read old versions with versionAsOf. DESCRIBE HISTORY, OPTIMIZE and VACUUM work.

Data

NYC TLC Trip Record Data, green taxi trips for January to March 2024, and the taxi zone lookup table.