Lesson 4 - Incremental & Resumable Pipelines

Welcome to Incremental & Resumable Pipelines#

In Lesson 3 you streamed a month of NYC taxi trips into a local SQLite warehouse a chunk at a time, never holding the whole file in memory. That works beautifully the first time. But a real pipeline does not run once — it runs again and again. January’s data lands, you load it. A month later February’s file appears, and you run the loader again. Then March. A scheduler will fire the exact same job every night, whether or not there is anything new to do.

Two things go wrong if you are naive about the repeat. First, wasted work: if the loader reprocesses January every single time February shows up, your nightly job gets slower and slower as history piles up, re-reading months that have not changed in weeks. Second, and far more dangerous, double-counting: if the loader runs twice for the same month — a retry after a network blip, an accidental double-click, a scheduler that fired twice — January’s ~2.96 million rows land in the table twice, and every total CityFlow’s dashboard reports is silently wrong.

Both problems have the same small, sturdy fix: a ledger. It is a tiny table that records which partitions (here, which months) are already loaded. Before touching a month, the loader asks the ledger; after a month fully lands, the loader writes it into the ledger — and that “mark done only after success” ordering is what makes a crashed run safe to restart. By the end of this lesson you will have a loader you could hand to a scheduler and trust.

By the end of this lesson, you will be able to:

• Explain why a repeatedly-run pipeline must be incremental (skip finished work) and idempotent (safe to re-run)
• Build a loaded_months ledger table alongside the trips data table
• Write a load_month function that skips already-loaded months and, for new ones, streams the parquet in batches and marks completion in the same transaction
• Prove re-running the whole pipeline does not change the row count — no double-counting
• Understand why “mark complete only after success” makes a crashed load resumable, and when you need a DELETE-before-insert guard

You only need the standard library’s sqlite3, plus pyarrow to stream the parquet. Let’s make the pipeline safe to repeat.

The Warehouse and the Ledger#

CityFlow keeps its trips in a local SQLite file. The full months live on the public TLC bucket; you download each one once and read the local copy by name:

# gate: skip
# Each month is a real ~50 MB parquet on the public NYC TLC bucket. Download once.
JAN = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-01.parquet"
FEB = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-02.parquet"

The design uses two tables, and the split is the whole idea. trips holds the data. loaded_months is the ledger: one row per month that has been fully loaded, with the row count and a timestamp so you have an audit trail. Create both in a fresh database, removing any leftover file first so re-running the lesson always starts clean:

import warnings; warnings.filterwarnings("ignore")
import os, sqlite3

DB = "incremental.db"
os.path.exists(DB) and os.remove(DB)          # start clean so re-running the lesson is safe

con = sqlite3.connect(DB)
con.execute("""
CREATE TABLE trips (
    tpep_pickup_datetime TEXT,
    PULocationID  INTEGER,
    DOLocationID  INTEGER,
    fare_amount   REAL,
    total_amount  REAL,
    month         TEXT
)""")
con.execute("""
CREATE TABLE loaded_months (
    month     TEXT PRIMARY KEY,
    rows      INTEGER,
    loaded_at TEXT
)""")
con.commit()

tables = con.execute(
    "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name").fetchall()
print("tables:", [t[0] for t in tables])
print("trips rows :", con.execute("SELECT COUNT(*) FROM trips").fetchone()[0])
print("ledger rows:", con.execute("SELECT COUNT(*) FROM loaded_months").fetchone()[0])
con.close()

tables: ['loaded_months', 'trips']
trips rows : 0
ledger rows: 0

Two empty tables. Notice two deliberate choices. The ledger’s month column is a PRIMARY KEY, so the database itself refuses to record the same month twice — a second insert of 2024-01 would raise rather than quietly duplicate the marker. And the trips table carries its own month column. You do not strictly need it yet, but it is what lets you say “everything belonging to February” in one WHERE clause, which turns out to matter for the robust resume pattern at the end of the lesson.

A Loader That Checks the Ledger First#

Now the function that does the real work. load_month takes the connection, a month string like "2024-01", and the path to that month’s parquet. Its logic is a gate followed by a load:

1. Ask the ledger. If the month is already recorded, print a skip message and return — touch nothing.
2. Otherwise stream it in. Read the parquet in batches (the out-of-core pattern from earlier in this module), insert each batch into trips, and — as the last write — record the month in the ledger. All of that happens inside a single with con: transaction, so either the whole month lands and gets marked, or nothing does.

import warnings; warnings.filterwarnings("ignore")
import sqlite3, time
import pyarrow.parquet as pq

KEEP = ["tpep_pickup_datetime", "PULocationID", "DOLocationID",
        "fare_amount", "total_amount"]

def load_month(con, month, path):
    # 1. Ask the ledger: is this partition already done?
    done = con.execute(
        "SELECT 1 FROM loaded_months WHERE month = ?", (month,)).fetchone()
    if done:
        print(f"skip {month} (already loaded)")
        return
    # 2. Not done -> stream the month in and mark it complete, all in ONE transaction.
    pf = pq.ParquetFile(path)
    rows_loaded = 0
    with con:                                  # commits at the end, rolls back on any error
        for batch in pf.iter_batches(batch_size=200_000, columns=KEEP):
            df = batch.to_pandas()
            df["tpep_pickup_datetime"] = df["tpep_pickup_datetime"].astype(str)
            df["month"] = month
            con.executemany(
                "INSERT INTO trips VALUES (?, ?, ?, ?, ?, ?)",
                list(df.itertuples(index=False, name=None)))
            rows_loaded += len(df)
        # the ledger marker is the LAST write in the transaction
        con.execute(
            "INSERT INTO loaded_months (month, rows, loaded_at) VALUES (?, ?, ?)",
            (month, rows_loaded, time.strftime("%Y-%m-%d %H:%M:%S")))
    print(f"loaded {month}: {rows_loaded:,} rows")

con = sqlite3.connect("incremental.db")

load_month(con, "2024-01", "yellow_tripdata_2024-01.parquet")
print("trips after Jan:", f"{con.execute('SELECT COUNT(*) FROM trips').fetchone()[0]:,}")

load_month(con, "2024-02", "yellow_tripdata_2024-02.parquet")
print("trips after Feb:", f"{con.execute('SELECT COUNT(*) FROM trips').fetchone()[0]:,}")

print("\nledger:")
for row in con.execute("SELECT month, rows, loaded_at FROM loaded_months ORDER BY month"):
    print(f"  {row[0]}  rows={row[1]:>9,}  loaded_at={row[2]}")
con.close()

loaded 2024-01: 2,964,624 rows
trips after Jan: 2,964,624
loaded 2024-02: 3,007,526 rows
trips after Feb: 5,972,150

ledger:
  2024-01  rows=2,964,624  loaded_at=2026-07-14 17:22:46
  2024-02  rows=3,007,526  loaded_at=2026-07-14 17:22:55

This is the incremental load working. The first call streamed January’s 2,964,624 rows in and the table held exactly that. The second call streamed only February’s 3,007,526 rows and the table grew to 5,972,150 — nearly six million trips, far more than you would want to hold in a single pandas frame. Crucially, the February call did not re-read January; it appended the new month and left the old one alone. The ledger now has two rows, each an honest record of what landed and when (the loaded_at timestamp is wall-clock, so it will differ on your machine — that is expected).

Idempotency: Running It Again Changes Nothing#

Now the property that makes this pipeline safe to schedule. Idempotency means running an operation twice has the same effect as running it once. Fire the exact same loader a second time — both months already in the ledger — and watch what happens to the row count:

import warnings; warnings.filterwarnings("ignore")
import sqlite3, time
import pyarrow.parquet as pq

KEEP = ["tpep_pickup_datetime", "PULocationID", "DOLocationID",
        "fare_amount", "total_amount"]

def load_month(con, month, path):
    done = con.execute(
        "SELECT 1 FROM loaded_months WHERE month = ?", (month,)).fetchone()
    if done:
        print(f"skip {month} (already loaded)")
        return
    pf = pq.ParquetFile(path)
    rows_loaded = 0
    with con:
        for batch in pf.iter_batches(batch_size=200_000, columns=KEEP):
            df = batch.to_pandas()
            df["tpep_pickup_datetime"] = df["tpep_pickup_datetime"].astype(str)
            df["month"] = month
            con.executemany(
                "INSERT INTO trips VALUES (?, ?, ?, ?, ?, ?)",
                list(df.itertuples(index=False, name=None)))
            rows_loaded += len(df)
        con.execute(
            "INSERT INTO loaded_months (month, rows, loaded_at) VALUES (?, ?, ?)",
            (month, rows_loaded, time.strftime("%Y-%m-%d %H:%M:%S")))
    print(f"loaded {month}: {rows_loaded:,} rows")

con = sqlite3.connect("incremental.db")

before = con.execute("SELECT COUNT(*) FROM trips").fetchone()[0]
print("trips before re-run:", f"{before:,}")

# Run the exact same pipeline a second time -- a scheduler would do this nightly.
load_month(con, "2024-01", "yellow_tripdata_2024-01.parquet")
load_month(con, "2024-02", "yellow_tripdata_2024-02.parquet")

after = con.execute("SELECT COUNT(*) FROM trips").fetchone()[0]
print("trips after re-run: ", f"{after:,}")
print("ledger still has", con.execute("SELECT COUNT(*) FROM loaded_months").fetchone()[0], "rows")
print("no double-counting:", before == after)
con.close()

trips before re-run: 5,972,150
skip 2024-01 (already loaded)
skip 2024-02 (already loaded)
trips after re-run:  5,972,150
no double-counting: True
ledger still has 2 rows

Both months printed skip ... (already loaded), not one row was inserted, and the total held at exactly 5,972,150. That is idempotency in one screen. Without the ledger, this second run would have appended January and February all over again, pushing the table to nearly twelve million rows and doubling every fare total on the dashboard. The ledger check turned a dangerous re-run into a cheap no-op — it read two rows from a tiny table and did nothing else.

This is exactly the safety you need before you let a scheduler anywhere near the job. A cron entry or an Airflow DAG that runs nightly will occasionally run twice — a manual backfill overlaps the schedule, a retry fires after a timeout that actually succeeded, someone reruns yesterday to be safe. With an idempotent loader, none of that can corrupt the warehouse.

The figure below shows the whole arrangement: incoming months meet the ledger gate, already-loaded months are skipped, and only a genuinely new month is ingested — then recorded, but only after it fully lands.

Resumability: A Crash Leaves the Ledger Honest#

Incremental loading solves waste; idempotency solves accidental repeats. The third payoff falls out of the same design almost for free: resumability. Because a month is recorded in the ledger only after it fully loads, a crash partway through a month leaves that month not marked — so a restart naturally picks up right where it stopped.

Watch it happen. The following block builds a fresh warehouse, loads January cleanly, then deliberately crashes partway through February to simulate a power loss or an OOM kill. The load runs inside with con:, so the exception rolls the whole February transaction back:

import warnings; warnings.filterwarnings("ignore")
import os, sqlite3, time
import pyarrow.parquet as pq

DB = "incremental.db"
os.path.exists(DB) and os.remove(DB)          # self-contained demo: start from an empty warehouse
con = sqlite3.connect(DB)
con.execute("""CREATE TABLE trips (
    tpep_pickup_datetime TEXT, PULocationID INTEGER, DOLocationID INTEGER,
    fare_amount REAL, total_amount REAL, month TEXT)""")
con.execute("""CREATE TABLE loaded_months (
    month TEXT PRIMARY KEY, rows INTEGER, loaded_at TEXT)""")
con.commit()

KEEP = ["tpep_pickup_datetime", "PULocationID", "DOLocationID",
        "fare_amount", "total_amount"]

class CrashError(Exception):
    pass

def load_month(con, month, path, crash_after=None):
    if con.execute("SELECT 1 FROM loaded_months WHERE month = ?", (month,)).fetchone():
        print(f"skip {month} (already loaded)")
        return
    pf = pq.ParquetFile(path)
    rows_loaded = 0
    with con:                                  # rolls back everything if we raise below
        for i, batch in enumerate(pf.iter_batches(batch_size=200_000, columns=KEEP)):
            df = batch.to_pandas()
            df["tpep_pickup_datetime"] = df["tpep_pickup_datetime"].astype(str)
            df["month"] = month
            con.executemany(
                "INSERT INTO trips VALUES (?, ?, ?, ?, ?, ?)",
                list(df.itertuples(index=False, name=None)))
            rows_loaded += len(df)
            if crash_after is not None and i == crash_after:
                raise CrashError(f"power lost while loading {month}")
        con.execute(
            "INSERT INTO loaded_months (month, rows, loaded_at) VALUES (?, ?, ?)",
            (month, rows_loaded, time.strftime("%Y-%m-%d %H:%M:%S")))
    print(f"loaded {month}: {rows_loaded:,} rows")

# January finishes cleanly and is marked done.
load_month(con, "2024-01", "yellow_tripdata_2024-01.parquet")

# February crashes after a few batches -- some rows were already inserted.
try:
    load_month(con, "2024-02", "yellow_tripdata_2024-02.parquet", crash_after=3)
except CrashError as e:
    print("CRASH:", e)

print("\n-- state right after the crash --")
print("trips rows:", f"{con.execute('SELECT COUNT(*) FROM trips').fetchone()[0]:,}")
print("ledger    :", con.execute("SELECT month FROM loaded_months ORDER BY month").fetchall())

# Restart the pipeline: January is skipped, February resumes from scratch and completes.
print("\n-- restart --")
load_month(con, "2024-01", "yellow_tripdata_2024-01.parquet")
load_month(con, "2024-02", "yellow_tripdata_2024-02.parquet")
print("trips rows:", f"{con.execute('SELECT COUNT(*) FROM trips').fetchone()[0]:,}")
print("ledger    :", con.execute("SELECT month FROM loaded_months ORDER BY month").fetchall())
con.close()

loaded 2024-01: 2,964,624 rows
CRASH: power lost while loading 2024-02

-- state right after the crash --
trips rows: 2,964,624
ledger    : [('2024-01',)]

-- restart --
skip 2024-01 (already loaded)
loaded 2024-02: 3,007,526 rows
trips rows: 5,972,150
ledger    : [('2024-01',), ('2024-02',)]

Read the state right after the crash carefully. February had inserted several batches before it failed — yet trips holds exactly 2,964,624 rows, January’s count and not one more, and the ledger lists only January. The with con: transaction rolled February’s partial inserts back completely, so the warehouse looks as if February never started. Then the restart does the obvious right thing with no special logic: it skips January (the ledger says done) and resumes February from the beginning, landing all 3,007,526 rows and finally marking it. The pipeline healed itself simply by re-running.

That is the entire payoff of “mark complete only after success.” Because the marker and the data commit together, there is no state in which a month is half-loaded and also recorded as done. A crash can only ever leave a month fully-done-and-marked or not-done-and-unmarked, and the ledger check handles both.

Practice Exercises#

Exercise 1 — Report what the ledger knows. After loading both months, write a query against loaded_months that prints each month, its row count, and the ledger’s grand total (SELECT SUM(rows) ...). Confirm the ledger’s total matches SELECT COUNT(*) FROM trips — the ledger should always agree with the data it gates. You should see 2,964,624 + 3,007,526 = 5,972,150.

Exercise 2 — Prove the re-run is a no-op. Capture SELECT COUNT(*) FROM trips into a variable, call load_month for both "2024-01" and "2024-02" again, capture the count a second time, and assert the two counts are equal. Then confirm the ledger still has exactly two rows. A passing assertion is your machine-checkable proof of idempotency.

Exercise 3 — Add the delete-before-insert guard. Write a variant reload_month(con, month, path) that does not consult the ledger to skip, but instead starts each load with con.execute("DELETE FROM trips WHERE month = ?", (month,)) and then re-inserts, using INSERT OR REPLACE for the ledger marker. Load January, then call reload_month for January again, and confirm the row count is unchanged — a different route to the same no-double-count guarantee, useful when a month’s data gets corrected and must be rebuilt.

Summary#

You turned a one-shot loader into a pipeline that is safe to run on a schedule forever. The key structure is a small ledger table, loaded_months, sitting beside the trips data table. load_month consults the ledger first: if a month is already recorded it prints skip and returns, so the pipeline is incremental — it never reprocesses a finished month, and adding February touched only February. For a new month it streams the parquet in batches and writes the ledger marker as the last statement of a single with con: transaction, so the month is recorded as done only when every one of its rows has committed. That gives idempotency: re-running the whole pipeline left the table at exactly 5,972,150 rows with no double-counting, because both months were simply skipped. And it gives resumability: a simulated crash midway through February rolled back cleanly, leaving trips at January’s 2,964,624 rows and February unmarked, so a restart resumed February and completed it with no special recovery code. For loaders that commit in pieces rather than one transaction per month, a DELETE FROM trips WHERE month = ? before inserting rebuilds a partition safely on retry — the same guarantee by another route.

Key Concepts#

• Ledger table — a tiny loaded_months(month PRIMARY KEY, rows, loaded_at) table that records which partitions are already loaded; the loader reads it to decide what to skip and writes it to mark completion.
• Incremental load — skip any month already in the ledger, so a nightly job processes only genuinely new data instead of reprocessing all of history every run.
• Idempotency — running the pipeline twice reaches the same state as running it once; the ledger’s skip check makes a re-run a cheap no-op, so re-runs and retries can never double-count.
• Mark complete only after success — writing the ledger marker as the final statement of the same transaction that inserts the data means a month is recorded as done exactly when it is done, never before.
• Resumability — because a crash leaves an unfinished month unmarked (rolled back by the transaction, or cleared by a DELETE-before-insert guard), simply re-running the loader resumes from the first month that never completed.

Why This Matters#

Every serious data pipeline runs on a schedule, and schedules mean repetition, retries, and the occasional crash. A loader that assumes it runs exactly once, exactly to completion, is a loader that will eventually corrupt its own warehouse — a duplicated month here, a half-loaded partition there, and the dashboard numbers quietly drift from the truth. The ledger pattern in this lesson is the small piece of engineering that makes repetition safe: it is why CityFlow can point a scheduler at this job and stop worrying about it. When you reach orchestration tools like Airflow later, you will not be learning idempotency from scratch — you will recognise it as the property you already built by hand, the one those tools quietly assume of every task they retry. Incremental, idempotent, resumable: three words that together mean “you can run this again, and it will be fine.”

Continue Building Your Skills#

You now have every ingredient of a production-grade batch loader: streaming a file too big for memory, accumulating results across chunks, a SQLite warehouse you can query, and a ledger that makes the whole thing incremental and resumable. Lesson 5 pulls the last thread — querying the warehouse efficiently with indexes so the summaries CityFlow’s dashboard needs come back fast even as the trips table grows past six million rows and beyond. Then the module’s guided project, Aggregate Months That Don’t Fit, is where all of it comes together into one pipeline: you will assemble chunked parquet ingestion, the numeric downcasting from Module 3, and the loaded_months ledger from this lesson into a single incremental loader, and use it to build a per-zone, per-day trip summary over several months of taxi data — far more than fits in RAM, processed safely one month at a time, and safe to re-run the moment a new month arrives.