Build an Incremental ETL Pipeline in Python Using PostgreSQL Watermarks

Incremental ETL avoids rereading an entire source table on every run. Instead, the pipeline remembers the latest source change it successfully loaded, then requests only records after that point. That remembered position is the watermark.

This tutorial builds a small but production-oriented pipeline that copies app.orders into warehouse.fact_orders. It uses PostgreSQL timestamps plus a primary-key tie-breaker as the watermark and performs idempotent upserts in Python.

Define the Watermark Contract First

A watermark is only reliable when its source column has a clear contract. In this example, each source row has:

• updated_at: when the application last changed the row
• id: a stable, unique primary key

The checkpoint is the pair:

(last_updated_at, last_id) 

Rows are ordered lexicographically:

(updated_at, id) 

For example, if the saved watermark is:

(2025-03-08 10:00:00+00, 42) 

the next extraction includes rows where:

(updated_at, id) > ('2025-03-08 10:00:00+00', 42) 

The id is important. Multiple rows can receive the same timestamp, especially when inserts happen in a batch. Without a tie-breaker, rows sharing the previous maximum timestamp may be skipped.

This design assumes:

1. updated_at is populated by the database or application for every insert and update.
2. updated_at is never intentionally moved backward.
3. Source rows are not physically deleted without another way to represent deletion.
4. The pipeline can safely reload a small overlap window.

The final assumption matters because timestamp watermarks are not a full change data capture system. We will revisit that limitation later.

Create Source, Target, and Pipeline-State Tables

For a small local example, use one PostgreSQL database with separate schemas. In a real system, the source and warehouse may be different databases, clusters, or services.

Run this SQL first:

CREATE SCHEMA IF NOT EXISTS app; CREATE SCHEMA IF NOT EXISTS warehouse; CREATE TABLE IF NOT EXISTS app.orders ( id BIGSERIAL PRIMARY KEY, customer_id BIGINT NOT NULL, status TEXT NOT NULL, total_cents INTEGER NOT NULL CHECK (total_cents >= 0), updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() ); -- Ensure application updates refresh the watermark column. CREATE OR REPLACE FUNCTION app.set_updated_at() RETURNS TRIGGER AS $$ BEGIN NEW.updated_at := clock_timestamp(); RETURN NEW; END; $$ LANGUAGE plpgsql; DROP TRIGGER IF EXISTS orders_set_updated_at ON app.orders; CREATE TRIGGER orders_set_updated_at BEFORE UPDATE ON app.orders FOR EACH ROW EXECUTE FUNCTION app.set_updated_at(); CREATE TABLE IF NOT EXISTS warehouse.fact_orders ( order_id BIGINT PRIMARY KEY, customer_id BIGINT NOT NULL, status TEXT NOT NULL, total_cents INTEGER NOT NULL, source_updated_at TIMESTAMPTZ NOT NULL, loaded_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() ); CREATE TABLE IF NOT EXISTS warehouse.pipeline_state ( pipeline_name TEXT PRIMARY KEY, last_updated_at TIMESTAMPTZ NOT NULL, last_id BIGINT NOT NULL, updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() ); INSERT INTO app.orders (customer_id, status, total_cents) VALUES (101, 'paid', 2599), (102, 'pending', 4999), (103, 'paid', 1200); 

The target table retains source_updated_at. That field is useful both for observability and for preventing an older reprocessed record from overwriting a newer target version.

Understand the Pipeline Architecture and Transaction Boundaries

The pipeline has four logical components:

PostgreSQL source table | | rows after saved watermark v Python extraction and transformation | | idempotent upsert v PostgreSQL warehouse table | | successful load only v PostgreSQL pipeline_state checkpoint 

The critical correctness rule is:

Advance the watermark only after the target rows have been written successfully.

If the process fails during loading, the saved checkpoint must not move. On the next execution, the same source rows are read again. That requires target writes to be idempotent: loading the same source row twice must produce the same final target state.

This example uses a transaction around target upserts and the state update. It also obtains a PostgreSQL advisory lock so two scheduler invocations cannot run the same pipeline concurrently.

The source extraction uses a REPEATABLE READ transaction. That gives the run a consistent source snapshot: the upper watermark and extracted rows describe the same view of the source table.

If you want a broader structured learning path after implementing this pattern, this data engineering roadmap is an optional next step.

Run the Incremental Python ETL Job

Install the PostgreSQL driver:

pip install "psycopg[binary]" 

Set a connection string. This example uses the same database for source and target:

export DATABASE_URL="postgresql://postgres:postgres@localhost:5432/analytics" 

Save the following as incremental_orders.py:

import os from datetime import datetime, timedelta, timezone import psycopg from psycopg.rows import dict_row DATABASE_URL = os.environ["DATABASE_URL"] PIPELINE_NAME = "orders_incremental_v1" # Re-read recent changes to tolerate slightly late commits. # Target upserts make this safe. LOOKBACK = timedelta(minutes=10) INITIAL_WATERMARK = datetime(1970, 1, 1, tzinfo=timezone.utc) def run_pipeline(): with ( psycopg.connect(DATABASE_URL, row_factory=dict_row) as source_conn, psycopg.connect(DATABASE_URL, row_factory=dict_row) as target_conn, ): # This transaction contains both target writes and state advancement.  with target_conn.transaction(): with target_conn.cursor() as target_cur: # Prevent concurrent runs of this exact pipeline.  target_cur.execute( "SELECT pg_advisory_xact_lock(hashtext(%s))", (PIPELINE_NAME,), ) target_cur.execute( """ SELECT last_updated_at, last_id FROM warehouse.pipeline_state WHERE pipeline_name = %s FOR UPDATE """, (PIPELINE_NAME,), ) saved_state = target_cur.fetchone() if saved_state: # Intentionally overlap prior work. id=0 means all IDs at  # the overlap timestamp range are eligible for reload.  lower_updated_at = saved_state["last_updated_at"] - LOOKBACK lower_id = 0 else: lower_updated_at = INITIAL_WATERMARK lower_id = 0 # Read a consistent source snapshot.  with source_conn.transaction(): with source_conn.cursor() as source_cur: source_cur.execute( "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY" ) # Capture an upper bound from the same source snapshot.  source_cur.execute( """ SELECT updated_at, id FROM app.orders ORDER BY updated_at DESC, id DESC LIMIT 1 """ ) upper_bound = source_cur.fetchone() if upper_bound is None: print("Source table is empty; nothing to load.") return upper_updated_at = upper_bound["updated_at"] upper_id = upper_bound["id"] source_cur.execute( """ SELECT id, customer_id, status, total_cents, updated_at AS source_updated_at FROM app.orders WHERE (updated_at, id) > (%s, %s) AND (updated_at, id) <= (%s, %s) ORDER BY updated_at, id """, ( lower_updated_at, lower_id, upper_updated_at, upper_id, ), ) rows = source_cur.fetchall() # Upsert makes retries and overlap reads safe.  if rows: target_cur.executemany( """ INSERT INTO warehouse.fact_orders ( order_id, customer_id, status, total_cents, source_updated_at ) VALUES ( %(id)s, %(customer_id)s, %(status)s, %(total_cents)s, %(source_updated_at)s ) ON CONFLICT (order_id) DO UPDATE SET customer_id = EXCLUDED.customer_id, status = EXCLUDED.status, total_cents = EXCLUDED.total_cents, source_updated_at = EXCLUDED.source_updated_at, loaded_at = clock_timestamp() WHERE warehouse.fact_orders.source_updated_at <= EXCLUDED.source_updated_at """, rows, ) # Advance state only after all target upserts succeeded.  target_cur.execute( """ INSERT INTO warehouse.pipeline_state ( pipeline_name, last_updated_at, last_id ) VALUES (%s, %s, %s) ON CONFLICT (pipeline_name) DO UPDATE SET last_updated_at = EXCLUDED.last_updated_at, last_id = EXCLUDED.last_id, updated_at = clock_timestamp() """, (PIPELINE_NAME, upper_updated_at, upper_id), ) print( f"Loaded or reprocessed {len(rows)} rows. " f"New watermark: ({upper_updated_at}, {upper_id})" ) if __name__ == "__main__": run_pipeline() 

Run it:

python incremental_orders.py 

Verify Expected Results and Incremental Behavior

After the first run, query the target:

SELECT order_id, customer_id, status, total_cents FROM warehouse.fact_orders ORDER BY order_id; 

Expected result:

The state table should contain one checkpoint:

SELECT pipeline_name, last_updated_at, last_id FROM warehouse.pipeline_state; 

Now simulate an update and a new order:

UPDATE app.orders SET status = 'paid' WHERE id = 2; INSERT INTO app.orders (customer_id, status, total_cents) VALUES (104, 'pending', 7500); 

Run the Python job again. The pipeline will:

• reread a small 10-minute overlap,
• update order 2 from pending to paid,
• insert the new order,
• advance the state to the newest (updated_at, id) pair.

Because the load uses INSERT ... ON CONFLICT DO UPDATE, overlap rows do not create duplicates.

Production Pitfalls That Watermarks Do Not Solve Automatically

Late commits can be missed without overlap. An application transaction may assign updated_at, remain open for several minutes, and commit after the ETL has already moved beyond that timestamp. The overlap window reduces this risk, but it is not a mathematical guarantee for arbitrarily long transactions.

Backdated timestamps break the contract. If an application sets updated_at from client-provided data or restores old timestamps during a migration, the row may fall behind the checkpoint and never be extracted.

Hard deletes are invisible. A query for changed rows cannot detect a row that no longer exists. Use soft deletes such as deleted_at, maintain a deletion tombstone table, or adopt change data capture (CDC).

Timestamp precision matters. PostgreSQL TIMESTAMPTZ supports microseconds, but a timestamp alone is still not a unique event identifier. The (updated_at, id) pair handles different rows with identical timestamps; it does not represent multiple distinct updates to the same row with the exact same timestamp.

Large batches need staging. executemany is appropriate for a small tutorial. For high volume, load into a staging table with PostgreSQL COPY, then perform a set-based INSERT ... SELECT ... ON CONFLICT.

For stronger guarantees across long-running transactions, deletes, and every row-level change, use PostgreSQL logical replication/CDC or an application outbox rather than relying solely on an updated_at watermark.

Test the Pipeline as a State Machine

Test more than whether rows appear in the target. The state transition is the important behavior.

Useful integration tests include:

1. Initial load: insert source rows, run the job, and assert target row count and state values.
2. Idempotent retry: run the job twice without source changes; assert no duplicate target rows.
3. Update propagation: update a source order, rerun, and assert the target reflects the new values.
4. Same-timestamp ordering: insert multiple rows with a controlled identical updated_at; confirm all are loaded because id breaks ties.
5. Failure before checkpoint: intentionally fail after extraction but before the state update, then rerun. Target data may be reprocessed, but no source change should be lost.
6. Concurrent execution: start two jobs and confirm the advisory lock serializes them.

Keep test data small and inspect warehouse.pipeline_state directly. A pipeline can look correct in the destination while silently holding an incorrect checkpoint that causes future data loss.

Takeaway

A PostgreSQL watermark pipeline is built on a simple rule: extract rows after a durable checkpoint, load them idempotently, and update that checkpoint only after the load succeeds. Use a composite (updated_at, id) watermark, a bounded source snapshot, transactional target writes, and an overlap window. It is a practical pattern for mutable tables, but for strict handling of deletes and long-running transactions, move to CDC or an outbox-based design.