Incremental, Idempotent & Orchestrated Pipelines

A real job runs every day. Sometimes it fails and runs again. Sometimes you run old days again on purpose.

This chapter makes a job safe to run more than once. It reads only new data, and it never stores the same row twice.

The source is the orders table of the app, with one extra column, updated_at:

source = [
    {"order_id": 101, "status": "paid", "amount": "40.00", "updated_at": "2026-01-03"},
    # ... orders 102 to 106 in the same shape ...
]

Data Pipeline chapter 12, Incremental Processing & Backfills, explains these ideas with SQL. This chapter builds them in Python.

A full refresh reads every row, every time. An incremental run reads only the rows that changed since the last run.

A watermark is the value that marks where the last run stopped. Here it is the newest updated_at that the job has read.

def read_full(source):
    return list(source)

def read_new(source, watermark):
    return [row for row in source if row["updated_at"] > watermark]

print(len(read_full(source)), len(read_new(source, "2026-01-21")))

The last run read up to 2026-01-21, so nothing is new yet. Now order 105 is paid on 2026-01-25:

source[4] = {"order_id": 105, "status": "paid", "amount": "22.00",
             "updated_at": "2026-01-25"}

print([row["order_id"] for row in read_new(source, "2026-01-21")])

A full refresh is simple and is often right for small tables. Incremental is cheaper for big tables, but it needs a watermark and more care.

A job is idempotent when running it twice with the same input gives the same result as running it once.

Suppose a run fails after it stored order 105, and the scheduler runs it again. If the job only appends rows, order 105 is stored twice:

batch = read_new(source, "2026-01-21")

target_list = []
target_list.extend(batch)
target_list.extend(batch)      # the second run
print(len(target_list))

Store the rows in a dict keyed by order_id instead. An upsert means update or insert. A known key replaces the old row, and a new key adds one.

def upsert(target, rows):
    for row in rows:
        target[row["order_id"]] = row

target = {}
upsert(target, batch)
upsert(target, batch)          # the second run
print(len(target), target[105]["amount"])

The upsert makes the job idempotent. Running it again is safe, so a retry needs no special care.

The watermark must survive between runs, so the job saves it in a file. A checkpoint is saved state like this, which the next run reads.

The order of the steps matters. Read the watermark, read the new rows, write them, and only then save the new watermark:

import json
import os

def read_watermark(path):
    if not os.path.exists(path):
        return "0000-00-00"          # first run: read everything
    with open(path, encoding="utf-8") as file:
        return json.load(file)["watermark"]

def save_watermark(path, value):
    with open(path, "w", encoding="utf-8") as file:
        json.dump({"watermark": value}, file)

def run(source, target, state_path):
    watermark = read_watermark(state_path)
    rows = read_new(source, watermark)
    upsert(target, rows)                       # 1. write
    if rows:                                   # 2. then save
        save_watermark(state_path, max(row["updated_at"] for row in rows))
    return len(rows)

target = {}
print(run(source, target, "state.json"), read_watermark("state.json"))
print(run(source, target, "state.json"), read_watermark("state.json"))
print(len(target))

If the job stops after the write but before the save, the next run reads the same rows again. The upsert makes that harmless.

If you saved the watermark first and the write then failed, the next run would skip those rows forever.

Late data is a row that reaches the source after a newer row was already read. Here a made-up order 186 was placed on 2026-01-24, but a slow system copies it in only on 2026-01-27.

Start from a watermark of 2026-01-21 again. The run on 2026-01-26 reads order 105, so the watermark moves to 2026-01-25. Then order 186 arrives:

target = {}
save_watermark("state.json", "2026-01-21")
print(run(source, target, "state.json"), read_watermark("state.json"))

source.append({"order_id": 186, "status": "paid", "amount": "30.00",
               "updated_at": "2026-01-24"})
print(run(source, target, "state.json"), 186 in target)

A replay window, or lookback, reads a few days before the watermark again. The upsert makes the repeated rows harmless:

from datetime import date, timedelta

def run_with_lookback(source, target, state_path, days=2):
    watermark = read_watermark(state_path)
    since = (date.fromisoformat(watermark) - timedelta(days=days)).isoformat()
    rows = [row for row in source if row["updated_at"] > since]
    upsert(target, rows)
    if rows:
        newest = max(row["updated_at"] for row in rows)
        save_watermark(state_path, max(watermark, newest))
    return [row["order_id"] for row in rows]

print(run_with_lookback(source, target, "state.json"), 186 in target, len(target))

max(watermark, newest) stops the watermark from moving back. Choose the window from how late your rows really arrive.

Many jobs write one file for each day, a partition (chapter 9). A rerun should replace that file, not add a second one.

A half-written file is also a danger. A reader may open it while the job is still writing. So write to a temporary file first, then rename it:

def paid_on(source, day):
    return [r["order_id"] for r in source
            if r["status"] == "paid" and r["updated_at"] == day]

def write_day(day, order_ids, folder="daily"):
    os.makedirs(folder, exist_ok=True)
    final = os.path.join(folder, f"{day}.json")
    temp = final + ".tmp"
    with open(temp, "w", encoding="utf-8") as file:
        json.dump(order_ids, file)
    os.replace(temp, final)
    return final

print(write_day("2026-01-25", paid_on(source, "2026-01-25")))
print(write_day("2026-01-25", paid_on(source, "2026-01-25")))
print(sorted(os.listdir("daily")), open("daily/2026-01-25.json").read())

os.replace renames the temp file to the final name, and replaces any old file with that name. On the same disk, the rename happens in one step.

So a reader sees the old file or the new file, never half of one. A rerun replaces the whole day, so it is idempotent.

A backfill runs the job again for days in the past. You do it on purpose, for example after you fix a bug or receive late data.

Order 186 arrived late, so the file for 2026-01-24 must be built. Loop over the days and call the same function:

for day in ["2026-01-24", "2026-01-25"]:
    write_day(day, paid_on(source, day))

print(sorted(os.listdir("daily")))
print(open("daily/2026-01-24.json").read(), open("daily/2026-01-25.json").read())

A backfill does not move the watermark. The watermark belongs to the normal daily runs.

A manifest is a small file that says what a run wrote. A reader checks for it before it reads the data.

Write the manifest last, after every data file. Then its presence means the run finished:

def write_manifest(run_date, days, watermark, folder="daily"):
    manifest = {"run_date": run_date, "days_written": days,
                "watermark": watermark}
    path = os.path.join(folder, f"_manifest_{run_date}.json")
    with open(path, "w", encoding="utf-8") as file:
        json.dump(manifest, file, sort_keys=True)
    return path

print(write_manifest("2026-01-27", ["2026-01-24", "2026-01-25"],
                     read_watermark("state.json")))
print(open("daily/_manifest_2026-01-27.json").read())

If the job fails before this step, there is no manifest. A reader then waits, instead of reading days that may be half done.

Later, the manifest also answers a question in one look: which days did that run write, and from which watermark?

A task is one step of a job, such as extract, transform or load. To extract is to read from the source, and to load is to write to the target.

An orchestrator is a tool that runs tasks in the right order, on a schedule, and retries them when they fail. Airflow, Prefect and Dagster are common orchestrators.

The order comes from the dependencies: transform needs extract, and load needs transform. Python can work out that order with graphlib:

from graphlib import TopologicalSorter

graph = {"transform": {"extract"}, "load": {"transform"}, "report": {"load"}}
print(list(TopologicalSorter(graph).static_order()))

Keep the work in normal Python functions, such as run and write_day. The orchestrator file should only call them. Then you can test each task without the orchestrator.

The Airflow knowledge base shows how Airflow describes this graph and runs it.

A retry runs a failed task again. It is only safe when the task is idempotent.

A hard case: the write succeeds, but the connection drops before the job hears that. The job thinks it failed and sends the same rows again.

Give each batch an id, and let the target remember the ids it has stored:

class Sink:
    def __init__(self):
        self.rows = []
        self.committed = set()

    def commit(self, batch_id, rows):
        if batch_id in self.committed:
            return "skipped"
        self.rows.extend(rows)
        self.committed.add(batch_id)
        return "written"

sink = Sink()
rows = [{"order_id": 105}, {"order_id": 186}]
print(sink.commit("2026-01-27", rows))
print(sink.commit("2026-01-27", rows))      # the retry
print(len(sink.rows))

In a real database, store the batch id in the same transaction as the rows. Then both are saved, or neither is.

Exactly once means each row ends up in the target one time, even when tasks fail and run again.

No orchestrator gives you this by itself. It comes from three choices in this chapter: write by key, save the watermark last, and remember batch ids.

Always say where the promise holds: in one target table, or in a whole chain of systems. Data Pipeline chapter 11, Idempotency, Retries & Exactly-Once, goes further.

A job that is safe to run again follows five rules:

1. Write by key, or replace a whole partition. Never only append.
2. Save the watermark after the write, not before.
3. Read a few days back, so late rows are not lost.
4. Pass the day in, so a backfill can run any day.
5. Give each batch an id, so a retry is skipped.

Each exercise opens in the Python workspace with its own data. Write the code, run it, and check the result against the tests.

Each one is about running the same job twice and getting the same result.