Incremental Transformation

Incremental transformation

Switch an existing @dlt.hub.transformation from write_disposition="replace" to stateful incremental loading, so only new or changed rows are processed on each run.

Requires: an existing transformation script from create-transformation. If you don't have one, run that skill first.

Reference: https://dlthub.com/docs/hub/transformations.md#incremental-transformations

Steps

1. Choose a pattern

Ask the user which situation applies if it is not obvious from the transformation.

2. Identify cursor column and Python type

The type annotation on the parameter (dlt.sources.incremental[T]) must match the column type in the destination.

3. Set write disposition and primary key

4. Add incremental parameter and apply to the query

Pattern 1 — Direct column

Use when aggregations or lookups produce a row per date/ID that should be upserted on each run.

from typing import Any
import pendulum
import dlt

@dlt.hub.transformation(write_disposition="merge", primary_key="date")
def crawl_counts_by_date(
    dataset: dlt.Dataset,
    crawled_at: dlt.sources.incremental[pendulum.DateTime] = dlt.sources.incremental(
        "date",
        initial_value=pendulum.datetime(2020, 1, 1, tz="UTC"),
        range_start="open",
    ),
) -> Any:
    yield dataset("SELECT date, COUNT(*) FROM connectors GROUP BY date").incremental(crawled_at)

Pattern 2 — Load-based (_dlt_loads.inserted_at)

Use when you want exactly the rows that arrived in new ingestion loads. dlthub joins _dlt_loads automatically — no manual JOIN needed in SQL or relation chains. initial_value is intentionally omitted here — the first run will process all existing loads; add initial_value=pendulum.now("UTC") to skip historical loads and start from now.

from typing import Any
import pendulum
import dlt

@dlt.hub.transformation(write_disposition="append")
def connectors_from_new_loads(
    dataset: dlt.Dataset,
    loaded_at: dlt.sources.incremental[pendulum.DateTime] = dlt.sources.incremental(
        "_dlt_loads.inserted_at",
        range_start="open",
    ),
) -> Any:
    yield (
        dataset.table("connectors")
        .incremental(loaded_at)
        .select("id", "name", "slug", "type", "_dlt_load_id")
    )

Pattern 3 — dltHub Platform scheduler

Use when the dltHub Platform scheduler sets the time window. initial_value sets the fallback start for the first run before the scheduler takes over.

from typing import Any
import pendulum
import dlt

@dlt.hub.transformation(write_disposition="append")
def orders_window(
    dataset: dlt.Dataset,
    window: dlt.sources.incremental[pendulum.DateTime] = dlt.sources.incremental(
        "created_at",
        initial_value=pendulum.datetime(2000, 1, 1, tz="UTC"),
        allow_external_schedulers=True,
        range_start="closed",
        range_end="open",
    ),
) -> Any:
    yield dataset.table("orders").incremental(window)

5. Rules and gotchas

• range_start="open" (default): excludes the last-seen cursor value on the next run, preventing double-processing the boundary row. Switch to "closed" only if the source may update the record exactly at the cursor boundary.
• merge requires primary_key: without it, rows are not deduped — merge silently behaves like append.
• lag (Optional): lag re-loads a trailing window so late-arriving rows are caught, but dlt infers the unit from the shape of the cursor value (detect_datetime_format) — a date ("2026-08-27") gets days, a datetime ("2026-08-27T00:00:00Z") gets seconds. "Last 7 days" is lag=7 on a date cursor, lag=604800 on a datetime one. Ref: https://dlthub.com/docs/general-usage/incremental/lag
• Load-based auto-join: the dotted path _dlt_loads.inserted_at triggers an automatic join to the _dlt_loads table. Do not write the JOIN yourself.
• columns= hints still apply: any column that may be NULL on the first incremental run needs an explicit columns= hint, same as in create-transformation.
• Same vs. different destination: when source and target use the same physical destination (same DB file or host), dlthub pushes SQL directly without materializing data. Across different destinations, dlthub streams Arrow chunks automatically.

6. Test incrementally

# First run — loads all data from initial_value
python transformations/<dataset_name>_to_cdm.py

# Inspect stored cursor state
uv run dlthub local pipeline info <pipeline_name>

# Second run — should process 0 new rows if no new data arrived
python transformations/<dataset_name>_to_cdm.py

Verify row counts with dlthub local pipeline show <pipeline_name> or the preview_table MCP tool. If the second run still reprocesses everything, check that primary_key is set and that the cursor column name matches exactly (case-sensitive).