Metadata-Driven Pipelines in Azure Data Factory

The first table takes an afternoon: a dataset, another dataset, a pipeline, a Copy activity. The second takes twenty minutes, by cloning the first. By the fortieth, the factory holds eighty datasets and forty pipelines that differ only in a table name, and a change to error handling has to be made forty times. Metadata-driven pipelines replace the forty with one.

The idea

Separate what to copy from how to copy it.

• The what is data: a control table with one row per table or folder, naming the source, the destination and how to load it.
• The how is one generic pipeline that reads those rows and does the same thing for each.

Adding a table becomes inserting a row. Microsoft makes the same case for the approach: you maintain the list of objects by updating the control table instead of redeploying pipelines, and you have a single place to see what is copied and how (Microsoft Learn: metadata-driven copy).

The building blocks

Four features of Azure Data Factory make the pattern possible:

Microsoft's bulk copy tutorial builds exactly this shape (Microsoft Learn: copy data in bulk). The skeleton:

{
  "name": "CopyAllFromControlTable",
  "properties": {
    "activities": [
      {
        "name": "GetObjects",
        "type": "Lookup",
        "typeProperties": {
          "source": {
            "type": "AzureSqlSource",
            "sqlReaderQuery": "SELECT SourceSchema, SourceTable, SinkFolder FROM etl.Control WHERE CopyEnabled = 1"
          },
          "dataset": { "referenceName": "ControlDb", "type": "DatasetReference" },
          "firstRowOnly": false
        }
      },
      {
        "name": "ForEachObject",
        "type": "ForEach",
        "dependsOn": [ { "activity": "GetObjects", "dependencyConditions": [ "Succeeded" ] } ],
        "typeProperties": {
          "items": { "value": "@activity('GetObjects').output.value", "type": "Expression" },
          "isSequential": false,
          "batchCount": 10,
          "activities": [
            {
              "name": "CopyOneObject",
              "type": "Copy",
              "inputs": [ {
                "referenceName": "SourceTableGeneric",
                "type": "DatasetReference",
                "parameters": {
                  "SchemaName": { "value": "@item().SourceSchema", "type": "Expression" },
                  "TableName": { "value": "@item().SourceTable", "type": "Expression" }
                }
              } ],
              "outputs": [ {
                "referenceName": "LakeFolderGeneric",
                "type": "DatasetReference",
                "parameters": { "FolderPath": { "value": "@item().SinkFolder", "type": "Expression" } }
              } ],
              "typeProperties": {
                "source": { "type": "SqlSource" },
                "sink": { "type": "ParquetSink" }
              }
            }
          ]
        }
      }
    ]
  }
}

Three expressions carry the whole design. @activity('GetObjects').output.value is the array of rows. @item() is the current row inside the loop. Inside each generic dataset, @dataset().TableName reads the parameter it was given. Parameterizing pipelines and Lookup and Get Metadata cover these in detail.

Note firstRowOnly set to false. Its default is true, which returns a single row and is the most common reason a metadata-driven pipeline copies one table and stops.

The pattern, working

The script below reproduces the pattern in SQLite on the Chinook sample database (Chinook on GitHub). A control table lists six tables: four loaded in full, one of which is switched off, and two loaded as deltas by a watermark column. One routine copies whatever its row describes. Between the two runs the source gains new invoices, and a seventh table is added by inserting one control row. It is a local reproduction, not Data Factory.

import shutil, sqlite3

shutil.copy("chinook.sqlite", "source.sqlite")              # a working copy stands in for the source system
db = sqlite3.connect(":memory:")                            # the destination, which also holds the control table
db.execute("ATTACH DATABASE 'source.sqlite' AS src")
db.executescript("""
    CREATE TABLE control (
        Id INTEGER PRIMARY KEY, SourceTable TEXT, SinkTable TEXT,
        LoadType TEXT CHECK (LoadType IN ('full', 'delta')),
        WatermarkColumn TEXT, WatermarkValue INTEGER DEFAULT 0,
        TaskId INTEGER DEFAULT 0, CopyEnabled INTEGER DEFAULT 1);
    INSERT INTO control (SourceTable, SinkTable, LoadType, WatermarkColumn, TaskId, CopyEnabled) VALUES
        ('Artist',      'stg_artist',       'full',  NULL,            0, 1),
        ('Album',       'stg_album',        'full',  NULL,            0, 1),
        ('Genre',       'stg_genre',        'full',  NULL,            0, 0),
        ('Customer',    'stg_customer',     'full',  NULL,            5, 1),
        ('Invoice',     'stg_invoice',      'delta', 'InvoiceId',     0, 1),
        ('InvoiceLine', 'stg_invoice_line', 'delta', 'InvoiceLineId', 0, 1);
""")


def copy_one(item):                                         # one generic "Copy activity", driven entirely by its row
    item_id, source, sink, load_type, column, old = item
    if load_type == "full":
        db.execute(f"DROP TABLE IF EXISTS {sink}")
        db.execute(f"CREATE TABLE {sink} AS SELECT * FROM src.{source}")
        return db.execute(f"SELECT COUNT(*) FROM {sink}").fetchone()[0]
    new = db.execute(f"SELECT MAX({column}) FROM src.{source}").fetchone()[0]
    db.execute(f"CREATE TABLE IF NOT EXISTS {sink} AS SELECT * FROM src.{source} WHERE 0")
    copied = db.execute(f"INSERT INTO {sink} SELECT * FROM src.{source} WHERE {column} > ? AND {column} <= ?",
                        (old, new)).rowcount
    db.execute("UPDATE control SET WatermarkValue = ? WHERE Id = ?", (new, item_id))
    return copied


def run_pipeline(label):
    items = db.execute("""SELECT Id, SourceTable, SinkTable, LoadType, WatermarkColumn, WatermarkValue
                          FROM control WHERE CopyEnabled = 1 ORDER BY TaskId DESC, Id""").fetchall()   # the Lookup
    print(f"{label}: {len(items)} objects")
    for item in items:                                                                               # the ForEach
        print(f"   {item[1]:12} {item[3]:6} {copy_one(item):5} rows")
    db.commit()


run_pipeline("run 1")

db.executescript("""
    INSERT INTO src.Invoice (InvoiceId, CustomerId, InvoiceDate, Total) VALUES (413, 5, '2014-01-05', 9.90), (414, 7, '2014-01-05', 1.98);
    INSERT INTO src.InvoiceLine (InvoiceLineId, InvoiceId, TrackId, UnitPrice, Quantity)
        VALUES (2241, 413, 1, 0.99, 10), (2242, 414, 2, 0.99, 1), (2243, 414, 3, 0.99, 1);
    -- onboarding a table is one row of metadata, and no change to the pipeline
    INSERT INTO control (SourceTable, SinkTable, LoadType) VALUES ('Employee', 'stg_employee', 'full');
""")
run_pipeline("run 2")

run 1: 5 objects
   Customer     full      59 rows
   Artist       full     275 rows
   Album        full     347 rows
   Invoice      delta    412 rows
   InvoiceLine  delta   2240 rows
run 2: 6 objects
   Customer     full      59 rows
   Artist       full     275 rows
   Album        full     347 rows
   Invoice      delta      2 rows
   InvoiceLine  delta      3 rows
   Employee     full       8 rows

Read the two runs side by side:

• Five objects, not six. Genre is in the control table and disabled, so the Lookup's query never returns it. Switching a load off is an update, not a deployment.
• Order follows the metadata. Customer has the highest TaskId and goes first.
• Full and delta share one routine. In run 2 the full tables are reloaded in their entirety, while Invoice and InvoiceLine copy only 2 and 3 new rows, because each row carries its own watermark and the routine updates it.
• Employee appears in run 2. Nothing in the pipeline changed. A row was inserted.

That last line is the argument for the pattern. The table names in such a routine come from the control table, so who may write to that table matters; treat it as code.

The limits that shape real designs

Two documented limits decide how the pattern scales.

Lookup returns at most 5,000 rows and 4 MB (Microsoft Learn: Lookup activity). A control table with more objects than that has to be read in pages.

ForEach runs at most 50 iterations at once, with the batch count as the upper limit (Microsoft Learn: ForEach activity). The loop cannot contain another ForEach, so nested iteration means calling a child pipeline. Control flow in ADF covers both points.

There is a third limit of a different kind. Not everything can be a parameter. Microsoft lists it among the known limitations of its own generator: the integration runtime name, the database type and the file format type cannot be parameterized. One generic pipeline can serve every SQL Server table; an Oracle source needs a second pipeline, though both can share one control table.

What the Copy Data tool generates

You do not have to build this by hand. The Copy Data tool has a metadata-driven copy task that produces the pipelines and the SQL scripts for the control tables. Seeing what it generates is instructive even if you later write your own.

Two control tables. The main control table has one row per object. Its columns hold the source settings, the sink settings and the copy settings as JSON, the loading behaviour, full or delta, and two columns used in the reproduction above: TaskId, which orders the objects, highest first, and CopyEnabled, 1 or 0. A second table holds connection settings when a parameterized linked service is used. A generated stored procedure writes the new watermark back after each delta load.

Three levels of pipeline.

The three levels are a direct answer to the two limits above. The default number of concurrent copy tasks is 20.

Updating it. The tool can reopen the control table for editing and produce a new SQL script. The pipelines are not redeployed; only the table changes.

Documented limitations. Besides the three properties that cannot be parameterized: the generated scripts use OPENJSON, so a SQL Server hosting the control table must be 2016 or later, and for storage sources only full load is offered, not incremental loading of new files.

When the pattern is worth it

The pattern suits ingestion, where the work really is the same for every table. It suits business transformation badly, where every table is different.

Trade-offs to accept knowingly

• The pipeline is harder to read. Every property is an expression. A newcomer cannot see what is copied without querying the control table.
• One run holds many loads. Monitoring shows one pipeline run with hundreds of activity runs. Per-table success and timing need to be recorded somewhere you can query.
• A bad row breaks at run time. A mistyped table name is not caught at publish.
• A failure needs a rerun plan. Rerunning the whole loop repeats the tables that succeeded unless the design tracks state.

Those last three are what a framework around the pattern is for: building a metadata framework for ADF pipelines covers the design. The delta logic is explained in incremental data loading.

Azure Data Factory and reusable pipeline design are covered in the Data Engineering course.

Sources

• Build large-scale data copy pipelines with metadata-driven approach in copy data tool (Microsoft Learn) — Microsoft Learn
• Copy data in bulk using Azure portal (Microsoft Learn) — Microsoft Learn
• Lookup activity in Azure Data Factory and Azure Synapse Analytics (Microsoft Learn) — Microsoft Learn
• ForEach activity in Azure Data Factory and Azure Synapse Analytics (Microsoft Learn) — Microsoft Learn
• Chinook sample database (GitHub) — GitHub