Building a Metadata Framework for ADF Pipelines

A Lookup, a ForEach and a parameterized Copy will load two hundred tables. Then one of them fails at 03:00, and the questions begin. Which one? Did the other 199 finish? Can it be rerun without reloading everything? Which tables loaded yesterday, and how many rows? A metadata framework is the part of the design that can answer.

This guide assumes the basic pattern from metadata-driven pipelines in Azure Data Factory and builds what goes around it.

What a framework adds

The fourth row is the one most home-grown frameworks leave out, and it is the one that makes reruns cheap.

Configuration tables

Microsoft's generated control table is a useful reference for what a row needs (Microsoft Learn: metadata-driven copy): source settings, sink settings, copy settings, a loading behaviour, an ordering value and an enabled flag, with connections in a table of their own. A hand-built design usually separates these further:

Source system. One row per connection: a name, a type, and the name of the secret that holds its credentials. Never the credential.

Object. One row per table or folder: source system, source name, target name, an enabled flag, and a stage number.

Load settings. How the object is loaded: full or delta, the watermark column and its current value, the key columns for a merge, partition settings for large tables.

Design rules that pay off:

• Typed columns for what you query, JSON for the rest. Microsoft's generator stores settings as JSON, which is flexible and unreadable in a query. Keep the fields you filter and join on, enabled, stage, load type, as real columns.
• Constraints on the configuration. A check constraint on the load type and a unique constraint on the target name catch mistakes when the row is inserted, not at 03:00.
• No secrets. Store the Key Vault secret's name.
• A validation query run at the start of every batch: enabled delta objects with no watermark column, duplicate targets, objects pointing at a disabled source.

Execution: orchestrator and workers

Keep the pipelines few and single-purpose:

• An orchestrator that opens a batch, loops over stages in order, and closes the batch.
• A stage pipeline that reads the objects still to run in that stage and loops over them in parallel.
• A worker per load pattern, such as database to lake, or lake to warehouse, that loads exactly one object and logs the result.

The split is forced by the product as much as by taste. A ForEach cannot contain another ForEach, so a loop over stages that contains a loop over objects has to call a child pipeline. The Execute Pipeline activity does that, and with wait on completion left at true the orchestrator waits for each stage before starting the next (Microsoft Learn: Execute Pipeline activity).

The worker receives its object's row as parameters and does three things: marks the object as started, runs the Copy activity, and records the outcome on both the success path and the failure path.

Logging

Data Factory keeps its own run history, and it is not shaped for the questions above: it is organised by pipeline and activity run, not by table. So the framework writes its own log.

What to record per object run:

The Copy activity returns its counts in its output, including rowsRead, rowsCopied, rowsSkipped and copyDuration (Microsoft Learn: monitor copy activity). A stored procedure activity after the copy reads them with expressions such as @activity('CopyOneObject').output.rowsCopied. The run ID, pipeline name and trigger name come from system variables such as @pipeline().RunId (Microsoft Learn: system variables). On the failure path, Microsoft's example for reading a failed copy's message is @activity('CopyOneObject').output.errors[0].Message (Microsoft Learn: how to send email).

Storing the pipeline run ID matters more than it looks. It is the link from a row in your log back to the full detail in the service's monitoring, when you need the activity's input and output.

Restart: the control logic

The built-in rerun helps, within limits. A pipeline can be rerun from a failed activity, and Microsoft documents how container activities behave when that happens: a ForEach always loops over the items it receives, and an Execute Pipeline activity triggers its child pipeline again (Microsoft Learn: visually monitor). What the service does not hold is your definition of done: which tables are complete for this business date.

The framework's answer is to make the list of work a question the log can answer: run the enabled objects of this stage that have not succeeded in this batch. A rerun then resumes by itself.

The script below reproduces that logic in SQLite, with Chinook tables as the sources (Chinook on GitHub). Five objects are configured in two stages, and one has a mistyped table name. It is a local reproduction of the control logic, 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:")
db.execute("ATTACH DATABASE 'source.sqlite' AS src")
db.executescript("""
    CREATE TABLE etl_object (ObjectId INTEGER PRIMARY KEY, SourceTable TEXT, SinkTable TEXT, Stage INTEGER, Enabled INTEGER DEFAULT 1);
    CREATE TABLE etl_batch  (BatchId INTEGER PRIMARY KEY, BusinessDate TEXT UNIQUE, Status TEXT);
    CREATE TABLE etl_log    (BatchId INTEGER, ObjectId INTEGER, Attempt INTEGER, Status TEXT,
                             RowsRead INTEGER, RowsWritten INTEGER, Error TEXT);
    INSERT INTO etl_object (SourceTable, SinkTable, Stage) VALUES
        ('Artist',      'raw_artist',       1),
        ('Album',       'raw_album',        1),
        ('Customer',    'raw_customer',     1),
        ('Invoices',    'raw_invoice',      1),      -- a typing mistake: the table is called Invoice
        ('InvoiceLine', 'raw_invoice_line', 2);
""")


def copy(source, sink):                                     # the worker: one generic copy, reporting what it did
    rows_read = db.execute(f"SELECT COUNT(*) FROM src.{source}").fetchone()[0]
    db.execute(f"DROP TABLE IF EXISTS {sink}")
    db.execute(f"CREATE TABLE {sink} AS SELECT * FROM src.{source}")
    return rows_read, db.execute(f"SELECT COUNT(*) FROM {sink}").fetchone()[0]


def run_batch(business_date):                               # the orchestrator
    found = db.execute("SELECT BatchId FROM etl_batch WHERE BusinessDate = ?", (business_date,)).fetchone()
    batch = found[0] if found else db.execute(              # one batch per business date: a rerun resumes it
        "INSERT INTO etl_batch (BusinessDate, Status) VALUES (?, 'Running')", (business_date,)).lastrowid
    for (stage,) in db.execute("SELECT DISTINCT Stage FROM etl_object WHERE Enabled = 1 ORDER BY Stage").fetchall():
        todo = db.execute("""
            SELECT ObjectId, SourceTable, SinkTable FROM etl_object o
            WHERE Enabled = 1 AND Stage = ? AND NOT EXISTS (
                SELECT 1 FROM etl_log l WHERE l.BatchId = ? AND l.ObjectId = o.ObjectId AND l.Status = 'Succeeded')
            """, (stage, batch)).fetchall()
        failed = 0
        for object_id, source, sink in todo:
            attempt = 1 + db.execute("SELECT COUNT(*) FROM etl_log WHERE BatchId = ? AND ObjectId = ?",
                                     (batch, object_id)).fetchone()[0]
            try:
                rows_read, rows_written = copy(source, sink)
                db.execute("INSERT INTO etl_log VALUES (?, ?, ?, 'Succeeded', ?, ?, NULL)",
                           (batch, object_id, attempt, rows_read, rows_written))
            except sqlite3.Error as error:
                failed += 1
                db.execute("INSERT INTO etl_log VALUES (?, ?, ?, 'Failed', NULL, NULL, ?)",
                           (batch, object_id, attempt, str(error)))
        print(f"   stage {stage}: {len(todo)} to run, {len(todo) - failed} succeeded, {failed} failed")
        if failed:                                          # later stages depend on this one: stop here
            db.execute("UPDATE etl_batch SET Status = 'Failed' WHERE BatchId = ?", (batch,))
            return print("   batch Failed; later stages not started")
    db.execute("UPDATE etl_batch SET Status = 'Succeeded' WHERE BatchId = ?", (batch,))
    print("   batch Succeeded")


print("run 1"); run_batch("2026-10-01")
db.execute("UPDATE etl_object SET SourceTable = 'Invoice' WHERE SourceTable = 'Invoices'")   # correct the metadata
print("run 2, same business date"); run_batch("2026-10-01")
print("run 3, same business date"); run_batch("2026-10-01")

print("\nthe log:")
for row in db.execute("""SELECT o.SinkTable, l.Attempt, l.Status, COALESCE(l.RowsRead, '-'), COALESCE(l.RowsWritten, '-'),
                                COALESCE(l.Error, '')
                         FROM etl_log l JOIN etl_object o USING (ObjectId) ORDER BY l.rowid"""):
    print("   {:17} attempt {}  {:9} read {:>4}  written {:>4}  {}".format(*row).rstrip())

run 1
   stage 1: 4 to run, 3 succeeded, 1 failed
   batch Failed; later stages not started
run 2, same business date
   stage 1: 1 to run, 1 succeeded, 0 failed
   stage 2: 1 to run, 1 succeeded, 0 failed
   batch Succeeded
run 3, same business date
   stage 1: 0 to run, 0 succeeded, 0 failed
   stage 2: 0 to run, 0 succeeded, 0 failed
   batch Succeeded

the log:
   raw_artist        attempt 1  Succeeded read  275  written  275
   raw_album         attempt 1  Succeeded read  347  written  347
   raw_customer      attempt 1  Succeeded read   59  written   59
   raw_invoice       attempt 1  Failed    read    -  written    -  no such table: src.Invoices
   raw_invoice       attempt 2  Succeeded read  412  written  412
   raw_invoice_line  attempt 1  Succeeded read 2240  written 2240

Follow the three runs:

• Run 1. Stage 1 has four objects; three load and one fails. The batch is marked failed and stage 2 is not started, because its object depends on stage 1 being complete.
• Run 2, after the configuration is corrected, for the same business date. Stage 1 has one object to run, not four. The three that succeeded are skipped. Then stage 2 runs and the batch succeeds.
• Run 3, the same date again, finds nothing to do in either stage. A scheduler that fires twice, or an engineer who reruns to be sure, causes no reload.

The log at the end is the audit trail: the failed attempt with its error text is kept beside the successful second attempt, and every success has matching read and written counts. That equality is a reconciliation check the framework gets for nothing; a difference means rows were skipped or rejected.

The key design decision is in one line: a batch is identified by its business date, so a rerun finds the existing batch instead of opening a new one. Idempotency in data pipelines explains why that property is worth designing for.

Ordering and dependencies

A stage number is the simplest dependency model that works: everything in stage 1 finishes before stage 2 starts, and objects within a stage run in parallel. It covers the common cases, reference data before transactions, raw before curated.

A full dependency graph, object A depends on B and C, is more precise and much more to build and debug. Add it only when stages genuinely cannot express the order.

Decide deliberately what a failure does. The reproduction stops at the end of a failed stage, after every object in it has been attempted. The alternative, continuing with objects that do not depend on the failure, needs the dependency graph.

Operating it

• Alert on the batch, not on each object. One notification per failed batch, listing the failed objects from the log. Monitoring and alerts in Azure Data Factory covers the alerting side, and handling errors and retries the failure paths inside the worker.
• Report from the log. Rows loaded per table per day, duration trends, the tables that fail most. These are plain queries.
• Keep configuration in source control. The control tables are part of the system. Hold their contents as scripts beside the pipelines and deploy both together; CI/CD for Azure Data Factory covers the pipeline side.
• Separate environments. The same pipelines, different configuration rows in development, test and production.
• Restrict who can change configuration. A row decides what is read and where it is written.

How far to go

Frameworks fail by growing. Guard against that:

• Start with three tables: objects, batches, log. Add a table when a real requirement needs it.
• Keep business logic out. The framework moves and tracks data. Transformations belong in SQL, data flows or notebooks that it calls.
• Resist configuring everything. If a setting has had the same value for every object for a year, it is not configuration.
• Consider the generated version first. For plain copying, the Copy Data tool's metadata-driven task gives you control tables and pipelines without design work. Build your own when you need logging, restart and ordering it does not provide.

Azure Data Factory and production 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
• Monitor copy activity (Microsoft Learn) — Microsoft Learn
• System variables supported by Azure Data Factory and Azure Synapse Analytics (Microsoft Learn) — Microsoft Learn
• Visually monitor Azure Data Factory (Microsoft Learn) — Microsoft Learn
• How to send email (Microsoft Learn) — Microsoft Learn
• Execute Pipeline activity in Azure Data Factory and Azure Synapse Analytics (Microsoft Learn) — Microsoft Learn
• Chinook sample database (GitHub) — GitHub