How to Implement Incremental Data Loading in an ETL Pipeline

The problem with full loads

A common ETL pattern looks like this: TRUNCATE the target, then copy everything from the source. It's simple, but it has two serious drawbacks:

• Performance degrades linearly - as the source grows, every run takes longer.
• Availability gap - between the TRUNCATE and the final COMMIT, the target table returns zero rows to any query that hits it.

Incremental loading fixes both by only moving rows that are new or changed.

Watermarks: remembering where you left off

A watermark is a saved timestamp (or sequence number) marking the last row you processed. You store it in a small pipeline_state table:

CREATE TABLE pipeline_state (
    pipeline_name TEXT PRIMARY KEY,
    last_watermark TIMESTAMPTZ,
    last_run_at   TIMESTAMPTZ
);

On each run, read the watermark, query only rows newer than it, process them, then advance the watermark to the maximum updated_at you saw.

Python implementation

import psycopg2
from datetime import datetime

DB_URL = "postgresql://postgres:postgres@localhost:5432/app"
PIPELINE_NAME = "products_sync"

def sync_products():
    conn = psycopg2.connect(DB_URL)
    cur = conn.cursor()

    # 1. Read the last watermark (first run defaults to epoch)
    cur.execute(
        "SELECT last_watermark FROM pipeline_state WHERE pipeline_name = %s",
        (PIPELINE_NAME,)
    )
    row = cur.fetchone()
    watermark = row[0] if row and row[0] else datetime(1970, 1, 1)

    # 2. Fetch only rows changed since the watermark
    cur.execute("""
        SELECT product_id, name, price, stock, updated_at
        FROM products_source
        WHERE updated_at > %s
        ORDER BY updated_at
    """, (watermark,))
    rows = cur.fetchall()

    # 3. Upsert each row - no TRUNCATE, target stays live
    new_watermark = watermark
    for pid, name, price, stock, updated_at in rows:
        cur.execute("""
            INSERT INTO products_target (product_id, name, price, stock, synced_at)
            VALUES (%s, %s, %s, %s, NOW())
            ON CONFLICT (product_id) DO UPDATE SET
                name      = EXCLUDED.name,
                price     = EXCLUDED.price,
                stock     = EXCLUDED.stock,
                synced_at = NOW()
        """, (pid, name, price, stock))
        if updated_at > new_watermark:
            new_watermark = updated_at

    # 4. Save the new watermark
    cur.execute("""
        INSERT INTO pipeline_state (pipeline_name, last_watermark, last_run_at)
        VALUES (%s, %s, NOW())
        ON CONFLICT (pipeline_name) DO UPDATE SET
            last_watermark = EXCLUDED.last_watermark,
            last_run_at    = NOW()
    """, (PIPELINE_NAME, new_watermark))

    conn.commit()
    cur.close()
    conn.close()
    print(f"Synced {len(rows)} rows; watermark now {new_watermark}")

Key design choices

• INSERT ... ON CONFLICT DO UPDATE (upsert) handles both new rows and updates to existing rows in a single statement - no separate UPDATE pass.
• Advance the watermark only after a successful commit - if the pipeline crashes mid-run, the next run re-processes the same window. This makes the pattern idempotent as long as upserts are idempotent.
• First-run bootstrap - defaulting to epoch (1970-01-01) means the very first run does a full load, which is fine. After that, every run is incremental.
• Use updated_at, not created_at - updated_at catches both inserts and updates. A watermark on created_at would miss rows that were changed after their initial insert.