One pipelinefor every table.

A pipeline per table works for three tables. At thirty, every change to the way you copy has to be made thirty times, and a new table means a new pipeline. A control table turns that around: the pipeline stays the same, and the list of what to copy lives in a table you can read, change and query.

#What goes in it

One row per source table, with what the pipeline needs to know about it:

• SchemaName and TableName: which table to copy.
• WatermarkColumn: the column that shows a row changed, usually a modified date. Lesson 3 uses it.
• LastWatermark: how far the last successful run got.
• IsActive: switch a table off without deleting its row.
• LastRunAt and LastRunStatus: what happened the last time, for you and for monitoring.

#Create it

Run this in the Azure SQL Database the pipeline will read from (the query editor in the Azure portal works). The etl schema keeps the pipeline's own tables apart from the business data.

CREATE SCHEMA etl;
GO

CREATE TABLE etl.LoadControl (
    SchemaName      sysname       NOT NULL,
    TableName       sysname       NOT NULL,
    WatermarkColumn sysname       NOT NULL,
    LastWatermark   datetime2     NOT NULL DEFAULT '1900-01-01',
    IsActive        bit           NOT NULL DEFAULT 1,
    LastRunAt       datetime2     NULL,
    LastRunStatus   nvarchar(20)  NULL,
    CONSTRAINT PK_LoadControl PRIMARY KEY (SchemaName, TableName)
);

#Fill it

With the AdventureWorksLT sample, three tables are enough to see it work. Each of them has a ModifiedDate column.

INSERT INTO etl.LoadControl (SchemaName, TableName, WatermarkColumn)
VALUES ('SalesLT', 'Customer',         'ModifiedDate'),
       ('SalesLT', 'Product',          'ModifiedDate'),
       ('SalesLT', 'SalesOrderHeader', 'ModifiedDate');

The watermark starts at 1 January 1900, so the first run copies everything. That is the full load; every run after it is incremental.

#Check it

SELECT SchemaName, TableName, WatermarkColumn, LastWatermark
FROM etl.LoadControl
WHERE IsActive = 1;

This query is the one the pipeline runs first, in the next lesson. Three rows back means three tables to copy.

The pipeline has three activities: a Lookup that reads the control table, a ForEach that goes through its rows, and a Copy inside the ForEach that copies one table. The trick that makes one Copy serve every table is a dataset that takes the table name as a parameter.

#Two parameterised datasets

In Data Factory Studio, create two datasets.

1. SourceTable, on the Azure SQL Database linked service. Add two parameters, SchemaName and TableName. Under Connection, choose to enter the table manually and set the schema to @dataset().SchemaName and the table to @dataset().TableName.
2. LakeFile, on the Data Lake Storage Gen2 linked service, in Parquet format. Add the parameters FolderPath and FileName, and set the file path to @dataset().FolderPath and @dataset().FileName, with a container of your choice (bronze here).

Parquet keeps the column types and compresses well, which matters once the files grow.

#The Lookup

Add a Lookup named GetTables on a dataset for the same database, with this query, and clear First row only, so it returns every row:

SELECT SchemaName, TableName, WatermarkColumn, LastWatermark
FROM etl.LoadControl
WHERE IsActive = 1;

#The ForEach

Add a ForEach named EachTable, connect it after GetTables on success, and set its items to the Lookup's rows:

@activity('GetTables').output.value

Leave Sequential off and set the batch count to 4, so four tables copy at the same time without crowding the source database.

#The Copy, once for every table

Inside the ForEach, add a Copy named CopyTable. item() is the current row of the control table.

• Source: the SourceTable dataset, with SchemaName set to @item().SchemaName and TableName to @item().TableName.
• Sink: the LakeFile dataset, with these two values:

FolderPath  @concat(item().SchemaName, '/', item().TableName)
FileName    @concat(item().TableName, '_', formatDateTime(utcNow(), 'yyyyMMddHHmmss'), '.parquet')

Every run writes a new file with the time in its name, so nothing is overwritten and you can see each run's output.

#Run it

Select Debug. The output shows one Lookup, one ForEach and three Copy runs, one per table, and the lake now has a folder per table:

bronze/SalesLT/Customer/Customer_20261022061500.parquet
bronze/SalesLT/Product/Product_20261022061500.parquet
bronze/SalesLT/SalesOrderHeader/SalesOrderHeader_20261022061500.parquet

Add a fourth row to the control table and debug again: four copies, with no change to the pipeline.

Right now every run copies every row. That is fine for a small table and slow for a big one. The watermark in the control table fixes it: each run copies only the rows changed since the last successful run.

#Find the new watermark first

Inside the ForEach, before the copy, add a Lookup named NewWatermark with First row only on. It reads the highest value of the watermark column now:

SELECT COALESCE(MAX(@{item().WatermarkColumn}), CAST('1900-01-01' AS datetime2)) AS NewWatermark
FROM @{item().SchemaName}.@{item().TableName}

Taking it before the copy matters. A row that changes while the copy runs gets a later date than this value, so the next run picks it up instead of losing it.

#Copy only the rows in between

Change the Copy's source from the table to a Query, and give it this dynamic content:

SELECT * FROM @{item().SchemaName}.@{item().TableName}
WHERE @{item().WatermarkColumn} > '@{formatDateTime(item().LastWatermark, 'yyyy-MM-dd HH:mm:ss.fffffff')}'
  AND @{item().WatermarkColumn} <= '@{formatDateTime(activity('NewWatermark').output.firstRow.NewWatermark, 'yyyy-MM-dd HH:mm:ss.fffffff')}'

The source dataset still needs its two parameters, but with a query they are not used.

#Move the watermark on, only on success

In the database, add a procedure that writes the result back:

CREATE PROCEDURE etl.SetWatermark
    @SchemaName sysname,
    @TableName  sysname,
    @Watermark  datetime2
AS
UPDATE etl.LoadControl
SET LastWatermark = @Watermark,
    LastRunAt     = SYSUTCDATETIME(),
    LastRunStatus = 'Succeeded'
WHERE SchemaName = @SchemaName
  AND TableName  = @TableName;

Then add a Stored procedure activity after the Copy, connected on success, that calls etl.SetWatermark with:

SchemaName  @item().SchemaName
TableName   @item().TableName
Watermark   @activity('NewWatermark').output.firstRow.NewWatermark

If the copy fails, the watermark stays where it was, and the next run copies the same rows again. Nothing is skipped.

#See it work

Run the pipeline twice. The first run copies everything, the second copies nothing, because nothing changed. Now change one row at the source:

UPDATE SalesLT.Customer
SET Phone = '555-0100', ModifiedDate = SYSUTCDATETIME()
WHERE CustomerID = 1;

Run again: the Customer copy moves one row, and the others none.

#What you built

A control table, one pipeline for every table in it, and incremental loads that never skip a row. Add a schedule trigger for every night and it runs with nobody watching.