Batch Pipelines and Time Windows

A batch pipeline runs on a schedule over a window of data. Almost every bug in one comes from confusion between the window it is supposed to process and the time it happens to run.

Wall-clock time is the bug

# scheduled at 02:00 every night
from datetime import date

def load_bad(con):
    today = date.today()
    con.execute("""
        create or replace table daily_revenue as
        select ordered_at, sum(amount) as revenue
        from orders where ordered_at = ?
        group by 1
    """, [today])

 ordered_at  revenue
 2026-01-05    41.20

At 02:00, “today” is two hours old. The dashboard shows a 98% collapse in revenue every morning, recovering through the day. Worse, re-running the job on Thursday to fix Tuesday processes Thursday.

Pass the window in instead:

# pipeline.py
import duckdb, sys
from datetime import date, timedelta

RUN_DATE = date.fromisoformat(sys.argv[1]) if len(sys.argv) > 1 else date.today() - timedelta(days=1)
con = duckdb.connect("bookshop.duckdb")

con.execute("begin transaction")
con.execute("delete from daily_revenue where ordered_at = ?", [RUN_DATE])
con.execute("""
    insert into daily_revenue
    select ordered_at, count(*), sum(amount)
    from orders
    where ordered_at = ? and status = 'completed'
    group by 1
""", [RUN_DATE])
con.execute("commit")

print(con.execute("select * from daily_revenue where ordered_at = ?", [RUN_DATE]).df().to_string(index=False))

python pipeline.py 2026-01-04

 ordered_at  orders  revenue
 2026-01-04       2    34.25

Now the run is a pure function of its parameter. Running it in March for a January date does the right thing, which is what makes backfills possible at all. Every orchestrator provides this value — Airflow’s {{ ds }}, Dagster’s partition key, a --date flag in a cron job.

The default is yesterday, not today: process the last complete window.

Closed-open intervals

con.execute("""
    select count(*) from orders
    where ordered_at between '2026-01-04' and '2026-01-05'
""").fetchone()

(5,)

con.execute("""
    select count(*) from orders
    where ordered_at >= '2026-01-04' and ordered_at < '2026-01-05'
""").fetchone()

(3,)

BETWEEN is inclusive at both ends, so consecutive daily runs both claim midnight and double-count it. Always write >= start and < end. With timestamps rather than dates the error is worse — between '2026-01-04' and '2026-01-05' on a timestamp column silently excludes everything after midnight on the 5th, so the last day of a range is nearly empty.

The high-water mark that loses rows

Date-partitioned filters need a partition column. When there is not one, pipelines track a mark:

con.execute("create or replace table watermark as select timestamp '2026-01-04 00:00:00' as last_seen")

def load_incremental():
    mark = con.execute("select last_seen from watermark").fetchone()[0]
    new = con.execute("select * from source_orders where updated_at > ?", [mark]).df()
    print(f"mark={mark} -> {len(new)} rows")
    if len(new):
        con.execute("insert into orders select * from source_orders where updated_at > ?", [mark])
        con.execute("update watermark set last_seen = (select max(updated_at) from orders)")

mark=2026-01-04 00:00:00 -> 3 rows

Now a row is created at 23:58 but reaches the source table at 00:04, after the run:

con.execute("""insert into source_orders values (1008, 1, date '2026-01-04', 'completed', 12.00,
                                                  timestamp '2026-01-04 23:58:00')""")
load_incremental()

mark=2026-01-04 23:59:12 -> 0 rows

Zero rows. Order 1008’s updated_at is below the mark, so it will never be selected again — permanently lost, with every run reporting success. This is the single most common silent data loss in batch pipelines.

Two fixes, and you usually want both:

def load_with_lookback(lookback_hours=48):
    mark = con.execute("select last_seen from watermark").fetchone()[0]
    since = mark - timedelta(hours=lookback_hours)
    con.execute("""
        merge into orders t
        using (select * from source_orders where updated_at > ?) s
        on t.order_id = s.order_id
        when matched then update set status = s.status, amount = s.amount
        when not matched then insert by name
    """, [since])
    con.execute("update watermark set last_seen = (select max(updated_at) from source_orders)")

load_with_lookback()
print(con.execute("select count(*) from orders").fetchone())

(4,)

The lookback re-reads a window wide enough to catch late arrivals, and the MERGE makes re-reading harmless. Size the window from the worst arrival delay you have actually observed, not the typical one.

Better still, when the source allows it: track ingestion time, which is monotonic by construction, rather than event time, which is not.

Backfills

# backfill.py
import subprocess
from datetime import date, timedelta

start, end = date(2026, 1, 1), date(2026, 1, 8)
day = start
while day < end:
    print(f"--- {day}")
    subprocess.run(["python", "pipeline.py", day.isoformat()], check=True)
    day += timedelta(days=1)

--- 2026-01-01
 ordered_at  orders  revenue
 2026-01-01       4    88.40
--- 2026-01-02
 ordered_at  orders  revenue
 2026-01-02       6   112.05
--- 2026-01-03
...
--- 2026-01-07
 ordered_at  orders  revenue
 2026-01-07       5    94.20

Seven independent runs, each idempotent. Three properties this gives you:

• Resumable. A failure on day 5 does not lose days 1-4.
• Parallel. Days do not depend on each other, so they can run at once.
• Retryable. Re-running day 3 replaces day 3 and touches nothing else.

Compare with the version that looks simpler:

con.execute("""
    insert into daily_revenue
    select ordered_at, count(*), sum(amount) from orders
    where ordered_at >= '2026-01-01' and ordered_at < '2026-01-08'
    group by 1
""")

One statement, and it has none of those three properties. Run it twice and January is doubled; kill it at 70% and you cannot tell what was written.

Bound the concurrency. Twenty parallel backfill tasks against a production database is an outage:

from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=4) as pool:
    list(pool.map(run_day, days))

Chunking a large window

def load_month_in_chunks(year, month, chunk_days=1):
    day = date(year, month, 1)
    while day.month == month:
        end = day + timedelta(days=chunk_days)
        n = con.execute("""
            select count(*) from source_orders
            where updated_at >= ? and updated_at < ?
        """, [day, end]).fetchone()[0]
        print(f"{day} .. {end}  {n:>6} rows")
        day = end

2026-01-01 .. 2026-01-02    4120 rows
2026-01-02 .. 2026-01-03    4488 rows
2026-01-03 .. 2026-01-04    3902 rows
...

A single query over a year either exhausts memory or holds a lock long enough to matter. Chunking by the same window the pipeline is partitioned on keeps each unit small, restartable, and observable — you can see progress rather than watching one query for forty minutes.

Making a run observable

con.execute("""
    create table if not exists pipeline_runs (
        pipeline varchar, run_date date, started_at timestamp,
        finished_at timestamp, rows_written bigint, status varchar
    )
""")

def run(run_date):
    con.execute("insert into pipeline_runs values ('daily_revenue', ?, now(), null, null, 'running')", [run_date])
    try:
        rows = load(run_date)
        con.execute("""update pipeline_runs set finished_at = now(), rows_written = ?, status = 'success'
                       where pipeline = 'daily_revenue' and run_date = ? and status = 'running'""", [rows, run_date])
    except Exception as e:
        con.execute("""update pipeline_runs set finished_at = now(), status = 'failed'
                       where pipeline = 'daily_revenue' and run_date = ? and status = 'running'""", [run_date])
        raise

print(con.execute("""
    select run_date, rows_written, status,
           datediff('second', started_at, finished_at) as seconds
    from pipeline_runs order by run_date desc limit 4
""").df().to_string(index=False))

   run_date  rows_written   status  seconds
 2026-01-07          4820  success       18
 2026-01-06          4712  success       17
 2026-01-05             0   failed        4
 2026-01-04          4688  success       19

Two things this catches that a green exit code does not: a run that succeeded with zero rows, and a runtime creeping upward. Both are how a pipeline degrades without ever failing.

Practice

python pipeline.py 2026-01-02

 ordered_at  orders  revenue
 2026-01-02       6   112.05

between:  5 rows
>= and <: 3 rows

mark=2026-01-04 23:59:12 -> 0 rows

--- 2026-01-01  ✓
--- 2026-01-02  ✓
--- 2026-01-03  ✓
^C

 ordered_at  orders  revenue
 2026-01-01       4    88.40
 2026-01-02       6   112.05
 2026-01-03       5    97.10

Next: partitioning and layout — how data is arranged on disk, and what that costs.