Incremental Data Loading in Azure Data Factory

The first version of most pipelines reloads everything, every night. It is simple and obviously correct, and it stops being acceptable when the table reaches tens of millions of rows or the business wants data every hour. Incremental loading, copying only what changed since last time, is the answer, and it is the point where pipelines become harder to get right.

Four approaches

Microsoft's documentation describes four ways to load incrementally with Data Factory (Microsoft Learn: incremental load overview):

The watermark is the one everybody starts with, so it is worth understanding thoroughly, including where it fails.

The watermark pattern

A watermark is the highest value of some column that the pipeline has already loaded. Each run loads the rows above it and then raises it. The documented solution has four parts (Microsoft Learn: incremental copy tutorial):

1. A watermark column in the source, such as a last-modified timestamp or an ever-increasing ID.
2. A place to store the watermark, typically a small control table.
3. Two Lookup activities: one reads the stored, old watermark; the other reads the new watermark, the current maximum of the column in the source.
4. A Copy activity for rows between the two, then a stored procedure that saves the new watermark.

In the Copy activity the source query takes both boundaries from the Lookup outputs. In the form used by Microsoft's tutorial:

SELECT * FROM dbo.Orders
WHERE LastModified >  '@{activity('LookupOldWatermark').output.firstRow.WatermarkValue}'
  AND LastModified <= '@{activity('LookupNewWatermark').output.firstRow.NewWatermarkValue}'

Two design points are built into that query.

The upper bound matters. Reading the new watermark first and using it as an upper limit fixes the window. Rows that arrive while the copy is running fall above the bound and are picked up next time, instead of being half-seen now and skipped later.

The watermark moves last. It is updated only after the copy succeeds. If the copy fails, the old watermark stands and the next run retries the same window.

The pattern, run on real data

The script below reproduces the four steps in SQLite against the Invoice table of the Chinook sample database (Chinook on GitHub), 412 rows, with a LastModified column added. It runs the load three times. Before the third run the source changes: two new invoices, one correction, one deletion, and one row that arrives late with an old timestamp. This is a local reproduction of the logic, not Data Factory.

import shutil, sqlite3

shutil.copy("chinook.sqlite", "work.sqlite")                 # never alter the downloaded file
db = sqlite3.connect("work.sqlite")
db.executescript("""
    ALTER TABLE Invoice ADD COLUMN LastModified TEXT;
    UPDATE Invoice SET LastModified = InvoiceDate;           -- pretend the source tracks changes
    CREATE TABLE sink_invoice (InvoiceId INTEGER PRIMARY KEY, CustomerId INTEGER, Total REAL, LastModified TEXT);
    CREATE TABLE watermark (TableName TEXT PRIMARY KEY, WatermarkValue TEXT);
    INSERT INTO watermark VALUES ('Invoice', '1900-01-01 00:00:00');
""")


def incremental_run(label):
    # Lookup 1: the old watermark.  Lookup 2: the new one, the highest value in the source right now.
    old = db.execute("SELECT WatermarkValue FROM watermark WHERE TableName = 'Invoice'").fetchone()[0]
    new = db.execute("SELECT MAX(LastModified) FROM Invoice").fetchone()[0]
    # Copy: only rows changed after the old watermark, up to and including the new one; upsert on the key
    copied = db.execute("""
        INSERT INTO sink_invoice
        SELECT InvoiceId, CustomerId, Total, LastModified FROM Invoice
        WHERE LastModified > ? AND LastModified <= ?
        ON CONFLICT (InvoiceId) DO UPDATE SET Total = excluded.Total, LastModified = excluded.LastModified
    """, (old, new)).rowcount
    # Stored procedure: move the watermark forward, only after the copy succeeded
    db.execute("UPDATE watermark SET WatermarkValue = ? WHERE TableName = 'Invoice'", (new,))
    db.commit()
    print(f"{label:28} window ({old[:10]}, {new[:10]}]  rows copied {copied:4}")


incremental_run("run 1: first load")
incremental_run("run 2: nothing changed")

db.executescript("""
    INSERT INTO Invoice (InvoiceId, CustomerId, InvoiceDate, Total, LastModified) VALUES
        (413, 5, '2014-01-05 00:00:00', 9.90, '2014-01-05 00:00:00'),
        (414, 7, '2014-01-05 00:00:00', 1.98, '2014-01-05 00:00:00');
    UPDATE Invoice SET Total = 99.00, LastModified = '2014-01-06 00:00:00' WHERE InvoiceId = 10;   -- a correction
    DELETE FROM Invoice WHERE InvoiceId = 20;                                                     -- a deletion
    INSERT INTO Invoice (InvoiceId, CustomerId, InvoiceDate, Total, LastModified) VALUES
        (415, 9, '2013-06-01 00:00:00', 5.94, '2013-06-01 00:00:00');                              -- arrives late, stamped in the past
""")
incremental_run("run 3: after source changes")

missing = [r[0] for r in db.execute("SELECT InvoiceId FROM Invoice EXCEPT SELECT InvoiceId FROM sink_invoice")]
extra = [r[0] for r in db.execute("SELECT InvoiceId FROM sink_invoice EXCEPT SELECT InvoiceId FROM Invoice")]
counts = db.execute("SELECT (SELECT COUNT(*) FROM Invoice), (SELECT COUNT(*) FROM sink_invoice)").fetchone()
print(f"source rows {counts[0]}, sink rows {counts[1]}")
print("in source but not in sink:", missing)
print("in sink but not in source:", extra)
print("invoice 10 total in sink:", db.execute("SELECT Total FROM sink_invoice WHERE InvoiceId = 10").fetchone()[0])

run 1: first load            window (1900-01-01, 2013-12-22]  rows copied  412
run 2: nothing changed       window (2013-12-22, 2013-12-22]  rows copied    0
run 3: after source changes  window (2013-12-22, 2014-01-06]  rows copied    3
source rows 414, sink rows 414
in source but not in sink: [415]
in sink but not in source: [20]
invoice 10 total in sink: 99.0

The first two runs behave perfectly. Run 1 loads all 412 rows. Run 2, with nothing changed, loads none, which is the whole point of the pattern.

Run 3 copies three rows: the two new invoices and the corrected one, whose total is now right in the sink. Then look at the comparison:

• Invoice 415 is in the source and not in the sink. It was inserted with a timestamp of June 2013, below a watermark that had already passed December 2013. No later run will ever pick it up.
• Invoice 20 is in the sink and not in the source. It was deleted at the source. A deleted row has no timestamp to exceed the watermark. It stays in the target for ever.
• The row counts match: 414 and 414. One missing row and one surplus row cancel out. A reconciliation by count alone would pass this load as correct.

What a watermark cannot see

The two failures above are the general limits of the pattern.

Late and backdated rows. A watermark assumes that new changes always carry a higher value than anything already loaded. That is false when rows are stamped by application servers with different clocks, when a long transaction commits after later ones, or when data is imported with historical timestamps.

Deletes. Nothing marks a row that no longer exists.

Updates that do not touch the column. If some code path changes a row without updating the timestamp, the change is invisible. The pattern is only as reliable as the discipline that maintains the column.

Mitigations, in rising order of effort:

• Overlap the window. Subtract a safety margin from the old watermark, an hour or a day, and upsert. Rows are reloaded harmlessly, and most late arrivals are caught. This is cheap and usually worth doing.
• Use a database-assigned value. A column the database itself sets on commit is harder to backdate than one supplied by an application.
• Soft deletes. Have the source mark rows as deleted and update the timestamp, instead of removing them.
• Periodic full reconciliation. Weekly, compare keys between source and target and repair differences. Compare keys or checksums, not counts.
• Change tracking or change data capture. Let the database report what changed, including deletes.

The alternatives

Change tracking. SQL Server and Azure SQL Database can record which rows were inserted, updated or deleted, and Data Factory has a documented pattern that reads those changes. It sees deletes and does not depend on a timestamp column. It requires the feature to be enabled on the source. CDC explained covers the idea behind it.

Tumbling windows instead of a stored watermark. If the source has a trustworthy timestamp, a tumbling window trigger can pass each window's start and end to the pipeline, which then needs no watermark table. The trigger keeps the state, and any window can be rerun or backfilled. It has the same blind spots for late rows and deletes. See tumbling window triggers.

Files by time-partitioned path. When files arrive in dated folders, load the folder for the period. Microsoft describes this as the most performant approach for incrementally loading new files, because nothing has to be scanned.

Files by last-modified date. The Copy activity can filter files by their last-modified time. It works for folders that are not date-partitioned, with a cost the documentation states plainly: the service still scans all the files, so a large folder takes a long time even when few files are new.

Building it in Data Factory

The pipeline has five activities:

Practical points:

• Upsert, never append. With an overlap or a retry, the same row can arrive twice. See the Copy activity guide.
• One control table for all tables, keyed by table name, driving a ForEach. Lookup and Get Metadata covers reading it.
• Index the watermark column at the source, or every run scans the table.
• Mind time zones and precision. Compare like with like; a watermark truncated to seconds against a column with fractional seconds will drop or repeat rows at the boundary.
• Keep the first run in mind. An initial watermark far in the past makes run 1 a full load, as in the example. For very large tables, do the initial load separately.
• Log rows copied per run. A sudden zero, or a sudden spike, is the earliest sign that the watermark column has stopped being maintained.

Making the whole thing safe to repeat is the underlying requirement; idempotency in data pipelines explains why.

Azure Data Factory, with incremental ETL design, is covered in the Data Engineering course.

Sources

• Incrementally load data from a source data store to a destination data store (Microsoft Learn) — Microsoft Learn
• Incrementally load data from Azure SQL Database to Azure Blob storage using the Azure portal (Microsoft Learn) — Microsoft Learn
• Lookup activity in Azure Data Factory and Azure Synapse Analytics (Microsoft Learn) — Microsoft Learn
• Create a trigger that runs a pipeline on a tumbling window (Microsoft Learn) — Microsoft Learn
• Chinook sample database (GitHub) — GitHub