Incremental Load Reference

Loading only what's new: watermarks, change data capture, file-based incremental ingestion, and merge/upsert patterns.

1. Full vs Incremental Load

Incremental load reduces redundant processing but trades away full-load's simplicity: every design below is really a strategy for answering one question — "how do we know, cheaply and reliably, what changed?"

2. Watermark-based Extraction

The classic database-side pattern: track a high-water-mark value (usually an updated_at timestamp or a monotonic ID) from the last successful run, and pull only rows past it next time.

-- Store the watermark in a small control table
SELECT last_loaded_at FROM etl_control WHERE source_table = 'orders';

-- Extract only rows changed since the watermark
SELECT * FROM orders
WHERE updated_at > :last_loaded_at
ORDER BY updated_at;

-- After a successful load, advance the watermark
UPDATE etl_control
SET last_loaded_at = :max_updated_at_this_run
WHERE source_table = 'orders';

Pitfalls: clock skew between the source and the ETL host, rows updated with a stale/unchanged timestamp, and hard deletes (a watermark on updated_at never sees a row that was removed). Use a slightly overlapping window and dedupe by key, or pair with CDC for deletes.

3. Incremental Extraction from APIs

SaaS and REST sources rarely expose a raw table to query — incremental pulls rely on whatever the API offers instead:

• Cursor / since parameter — pass the last-seen ID or timestamp back as a query parameter (?updated_since=...); the same watermark idea as a database pull.
• Pagination tokens — APIs that page results return an opaque next_cursor; persist it so a resumed job continues instead of restarting from page one.
• Webhooks — the source pushes a change event as it happens, avoiding polling entirely, at the cost of needing a durable receiver (queue or endpoint) that never drops an event.
• Rate limits — incremental pulls should shrink request volume, but a badly chosen page size can still throttle; back off and retry with jitter rather than failing the whole run.

import requests, json

state = json.load(open("state.json"))  # {"cursor": "...", "updated_since": "2026-07-28T00:00:00Z"}

resp = requests.get(
    "https://api.example.com/orders",
    params={"updated_since": state["updated_since"], "cursor": state["cursor"]},
    headers={"Authorization": f"Bearer {TOKEN}"}
)
data = resp.json()
process(data["records"])

state["cursor"] = data.get("next_cursor")
state["updated_since"] = data["records"][-1]["updated_at"] if data["records"] else state["updated_since"]
json.dump(state, open("state.json", "w"))

4. Incremental Ingestion from Files

When the source is files landing in object storage rather than a live database, the question becomes "which files haven't we loaded yet?" Databricks offers two built-in mechanisms:

-- COPY INTO: run repeatedly, only new files are picked up
COPY INTO my_table
FROM '/path/to/files'
FILEFORMAT = CSV
FORMAT_OPTIONS ('delimiter' = '|', 'header' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

# Auto Loader: a streaming source that discovers new files as they land
spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .option("cloudFiles.schemaLocation", "/schema/orders") \
    .load("/path/to/files") \
  .writeStream \
    .option("checkpointLocation", "/checkpoints/orders") \
    .option("mergeSchema", "true") \
    .trigger(availableNow=True) \
    .table("orders")

Rule of thumb: reach for COPY INTO on modest, occasional batches; reach for Auto Loader once file counts climb into the millions or you need near-real-time pickup. trigger(availableNow=True) runs Auto Loader as a one-shot batch over everything new, rather than continuously.

5. Change Data Capture

CDC reads database changes directly from the transaction log instead of polling with a watermark query — it captures inserts, updates, and deletes, in order, with minimal load on the source.

• Log-based CDC (Debezium, AWS DMS, Kafka Connect) — tails the database's write-ahead/binlog and emits a change event per row mutation.
• Trigger-based CDC — database triggers write changes into a shadow table; simpler to set up, adds write overhead to the source.
• Delta Lake Change Data Feed — once enabled on a Delta table, every write automatically exposes a queryable stream of row-level changes.

-- Enable Change Data Feed on a Delta table
ALTER TABLE orders SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- Read only what changed between two versions
SELECT * FROM table_changes('orders', 10, 15)
WHERE _change_type IN ('insert', 'update_postimage', 'delete');

6. CDC Tooling Compared

Log-based CDC (Debezium, DMS) is preferred when available: it adds near-zero load to the source and captures deletes natively. Reach for connector-based tools when the source is a SaaS API with no accessible log.

7. Delivery Semantics & Idempotency

• At-most-once — a message may be lost, never duplicated. Rarely acceptable for analytical loads.
• At-least-once — nothing is lost, but retries after a partial failure can redeliver the same record. The common default for streaming systems (Kinesis, Kafka).
• Exactly-once — achieved not by magic delivery but by pairing at-least-once delivery with an idempotent write: dedupe by a natural or surrogate key (a MERGE/upsert, or a unique constraint) so replays are harmless.

Practical idempotency levers: a deterministic primary key on every record (not an auto-increment assigned at write time), a MERGE/upsert instead of a blind INSERT, and checkpoints that record what was durably committed to the target — not just what was read from the source.

8. Late Data & Streaming Watermarks

In a continuous pipeline, "incremental" means handling data that arrives out of order, not just data that arrives late relative to wall-clock time. Structured Streaming's watermark tells the engine how long to keep waiting for stragglers before finalizing a window:

from pyspark.sql.functions import window

events \
  .withWatermark("event_time", "15 minutes") \
  .groupBy(window("event_time", "5 minutes"), "device_id") \
  .count()

A tighter watermark finalizes results sooner but drops more genuinely-late events; a looser one keeps more state in memory and delays output. Size it against how late your source realistically arrives, not an arbitrary default.

9. Merge / Upsert

Once changed rows are identified, they're applied to the target with an upsert — insert new keys, update existing ones, optionally delete removed ones — rather than blindly appending duplicates.

-- Delta Lake MERGE INTO
MERGE INTO target AS t
USING changes AS s
ON t.order_id = s.order_id
WHEN MATCHED AND s._change_type = 'delete' THEN DELETE
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

-- Redshift equivalent: stage, then merge with a delete+insert
DELETE FROM target USING staging s WHERE target.order_id = s.order_id;
INSERT INTO target SELECT * FROM staging;

10. Partitioned Incremental Loads

Partitioning the target by load date (or event date) turns most incremental writes into an overwrite of just the affected partitions, instead of a row-level merge — simpler and often faster when a source naturally arrives in daily batches.

df.write \
  .mode("overwrite") \
  .option("partitionOverwriteMode", "dynamic") \
  .partitionBy("load_date") \
  .parquet("/data/orders/")

Dynamic partition overwrite replaces only the partitions present in the incoming DataFrame, leaving every other date's data untouched — a much cheaper substitute for a row-by-row merge when the grain of change is "a whole day changed," not "a few scattered rows changed."

11. Backfill & Reprocessing

Incremental pipelines still need a path back to "reload everything" — a source fixes a bug retroactively, a bad job corrupts a partition, or business logic changes and history must be recomputed.

• Parameterize the job by date range from the start, so a backfill is "run the same code for a wider range" rather than a special one-off script.
• Reprocess into a new location, validate, then swap/rename — safer than overwriting the live table in place.
• Reset the watermark or checkpoint deliberately and expect duplicates downstream; make sure the merge/upsert step is idempotent (§7) so replays don't double-count.

12. Monitoring & Data Quality

• Freshness — alert when the max loaded timestamp falls too far behind wall-clock time; a silent incremental job that stops finding "new" rows is easy to miss otherwise.
• Row-count sanity checks — a run that loads zero rows, or wildly more than a typical run, usually signals a broken watermark or an upstream outage rather than a genuinely quiet day.
• Duplicate / key-collision checks — periodically verify the primary key is still unique in the target; a leaking idempotency guarantee shows up here first.
• Checkpoint/watermark visibility — log the watermark value or offset committed each run so a stalled pipeline is diagnosable from history, not just its current state.

13. Design Checklist