The Modern Data Engineer Part 16: Metadata-Driven Pipelines

A loop set to thread = rand_int() % 1

At a previous client I spent a year keeping an ingestion platform running with one colleague, after the consultancy that built it on Databricks and Azure had withdrawn and the client couldn’t find anyone able to operate it. The nightly load started at midnight and ran six hours at a minimum, with spikes to ten or twelve, so nobody could promise fresh data the next morning. The parallel loop was a SQL statement that took a random number, applied a modulus, and used the remainder to assign each load to one of a handful of Azure Data Factory threads. At some point the assignment had been set to thread = rand_int() % 1, which put every load on thread 0, and since nobody left at the client understood the loop, it stayed that way for longer than anyone could say. That switched parallelism off, and it was one of many things adding to the run time, but every night’s data still got there eventually. Nobody fixed it because of that, and because finding out would have cost a week the two of us didn’t have.

What the control-database design looked like

The loop was one symptom of how the platform was built. The metadata lived on a SQL Server. A set of input tables held one row per source, and a stored procedure read those rows and assembled the select statement that ADF’s copy activity would run. Extraction, load and whatever transformation happened on the way all lived as dynamic SQL inside procedures that called views that called functions, and nothing in any of it handled a row deleted at the source. To know what actually ran for one source on one night you had to execute the procedure and read the string it produced.

Whoever built it had been ambitious about automation and never got to finish it. One table on that SQL Server held every column of every source, with its name, its type and a flag for whether it was part of the primary key the incremental logic used, and that table was dropped on every run. For SQL sources the tool then scraped the source’s own metadata and refilled it. It had no way to do that for API sources, so an insert script ran straight after the drop to append their columns by hand. Update statements followed for columns that should have been part of the key but weren’t marked as such at the source, since nothing in the source said so. All of it had to run again after every drop. The scraping ran in parallel against that one database, and a substantial share of the nightly failures started there, as deadlocks and transient availability errors on the metadata server, before a single row had been extracted from any source.

The tables were half under source control, as a Visual Studio project without real CI, so changing a table and getting the change into an environment was a nightmare every time. Part 1 named version control, testing and CI/CD as the baseline for the job, and this platform had version control and neither of the others.

How ADF reports a run’s status was the last problem. ADF derives a run’s status from its leaf activities only, so a copy activity that failed, then ran its on-failure handler successfully, counted as a success. With pipelines nested inside pipelines the outer run showed green while the inner runs had to be opened one by one to find the red. Retries were configured with long fixed intervals so a source would have time to recover. A step that failed past its retries left no way to reschedule it, so every run waited through the full retry interval and the failure still had to be found by hand.

Why the logic ended up in the database

These symptoms have one cause, which is where the logic had been put. Part 7 called ADF and Glue orchestration runtimes that automate the extract, and that’s the job they’re good at. ADF has no place to hold the extraction, load and transformation logic for every source. So the logic moved out of ADF into the SQL Server that held the metadata, one weekly fix at a time.

The fix we landed on after a year kept ADF, but only as a scheduler: a time trigger that scans a directory for spec files and hands each one to Databricks compute, and nothing else lives in ADF. The logic moved into a Python package released as a versioned wheel, driven by one YAML spec file per source under source control, with the YAML deployed by a pipeline. Replacing ADF outright would have been worse, since you want a scheduler anyway. The replacement is four objects and the sequence that connects them. A spec file per source holds what differs between sources, and a thin engine holds the behaviour they share. Two columns on every Bronze table, a load timestamp and a run id, record what landed, and a run record says what happened.

The spec file, which the series already has

The spec file isn’t new here. Part 9 introduced a per-source spec file that a home-grown extractor reads, declaring which columns are PII and how to generalize them. Part 14 relied on it as the file a privacy officer audits. Part 15 said the description ships in the same file as the classification tags. This is that file, with the load settings added.

Per source, the file holds the source, the load pattern (full load, watermark, CDC or partition reload), the key and delta columns where the pattern needs them, the contract, an owner, an availability window, and the descriptive fields Part 15 listed. The contract section can follow ODCS, the standard Part 10 recommended, and doesn’t have to. The schema and the quality rules do have to sit in the file the engine reads, so there’s no second document to keep in sync. Abridged, a spec looks like this.

source: crm.customers
owner: crm-platform-team
available_by: "06:00"
description: One row per customer as the CRM holds it
classification: {domain: sales, sensitivity: internal}
load_pattern: watermark
primary_key: [customer_id]
delta_columns: [modified_at]
contract:
  schema:
    - {name: customer_id, type: string, required: true}
    - {name: status, type: string, required: true}
  quality:
    - {rule: status in [active, churned, prospect], on_fail: quarantine}
    - {rule: modified_at not null, on_fail: quarantine}

The primary_key and delta_columns above are declared once, in a file under version control, and nothing rescrapes them. The file never holds a column mapping. Part 10 showed that an alias map inside the loader leaves Bronze holding the renamed columns instead of what the source sent, so anything that renames or reshapes belongs downstream in the source-aligned Silver model, which is also why the spec above has no mapping section.

The engine

A file on its own does nothing, so the second object is the engine: a thin Python package, released as a versioned wheel and installed on the compute the scheduler hands files to. It reads one spec at a time and executes the load pattern the spec names. Every difference between two sources is a difference between their spec files. Part 10’s story about a nightly export whose schema changed happened on this same platform, and the extraction framework mentioned there is this engine.

It runs the contract at run time. The structural check compares what arrived against the schema section, in front of Bronze, and the value-level rules read the data. Parts 10 and 11 both listed tools that could run the checks and did not pick one. Here the engine runs them as part of the load, before anything gets marked complete. Part 2 argued for schema validation in the platform as a runtime contract check and out of CI, and here it runs in the engine.

What passes lands in Bronze exactly as the source sent it. Where the load pattern detects deletes, the engine also writes the raw-layer deletion registry from Part 6, as target data in Bronze, the same way it writes the load timestamp and run id onto every row. In the lived version that was a shadow table per source holding only the primary-key columns, refreshed every day and compared against the source to find the rows that had disappeared, and that’s the other reason the spec has to declare primary_key.

The package has a command line that takes overrides. In practice we used one override, forcing a full run for a source whose spec says delta, when we suspected a slice had missed rows. The override is a run-time argument, so the spec stays the single description of the source and the exception shows up in the run record instead of as a flag in a control table that somebody has to remember to reset. The package also schedules its own retries dynamically. The fixed wait the old design put before each retry goes away. Unit tests run on a laptop, which Part 2 called for when it named cluster-bound tests a smell.

Where the state lives

Every Bronze table got two columns the engine writes: a load timestamp and a run id. The engine never wrote rows straight into Bronze. It wrote the extracted slice, with those two columns on every row, to a temporary folder first. Then it moved the finished file into Bronze with a command-line copy in one atomic step, so a half-recorded slice never appeared in Bronze. Keeping the slice on disk instead of in Python memory also meant a large extract didn’t overload the process. The next delta starts from the highest load timestamp among run ids that the run record marks complete. A run that landed its slice but was never marked complete has rows in Bronze under an incomplete run id, so the next run skips them when it works out its starting point. There’s no watermark table and no last-run row, since the resume point is whatever Bronze already contains. So the engine keeps no persistent control state of its own. Everything it needs for the next run is already in Bronze.

The refinement I’d make today is a manifest, a small object the run publishes to the target only when it completes, holding which slice landed, its bounds (watermark value, partition or snapshot time) and when. It gives the next run a single small record of what exists, read without querying Bronze, and with that the design works on plain object storage with no queryable Bronze at all, since the copy already handles the atomic landing. We didn’t have it at that client, and I’d add it now.

The Bronze columns and the run record, or the manifest in the refined version, also give you two of the things Part 15 said a data agent needs: freshness and lineage. Freshness is the time the last completed slice landed, compared with the availability window in the spec. Lineage is the source and the run id behind every Bronze row. That is lineage at the level of a whole slice, coarser than the column-level lineage Unity Catalog records on its own, which Part 5 covered, and it does not extend past Bronze. Both are queries over the run records, and nobody has to maintain them, since every run writes its record whether anyone asks for it or not.

The run record

Where Bronze’s columns record what landed, the run record covers what happened during the run, including the nights when nothing did, and it’s the completion marker the delta derivation depends on. The engine writes one per source per run, whether or not it succeeded and whether or not any consumer ever reads it. Ours held the run id, a completion status, counts of rows extracted, transformed, filtered and loaded, and the duration. I’d add error text and the spec and engine versions today.

A failed step leaves the record with no completion status. ADF’s status model stops mattering, because ADF no longer knows anything about the load beyond having handed over a file, and whether last night’s customers load worked is one row to look at. With the availability window and the owner from the spec it can also say whose failure it was, the platform’s or the source’s, which the first design never could.

The run lifecycle

The scheduler fires on its time trigger, scans the directory of spec files and hands each file to compute, which is its whole job.

The engine reads the spec, derives the starting point from Bronze (or from the last manifest, in the refined version), checks whether the source is inside its availability window, and extracts according to the load pattern. The extractor writes its whole output to a temporary folder that sits outside the warehouse, in front of Bronze, which is Part 11’s staged write. The contract checks run against the slice there.

On a tripped check, the slice stays in quarantine outside the warehouse, ingestion for that source circuit-breaks, no completion is recorded, the run record says which rule tripped, and the layer owner on the warehouse side is paged. On success, the finished file is copied into Bronze in one step, as sent, with its load timestamp and run id on every row, the deletion registry is written where the pattern detects deletes, and the run record is written with its completion status last. Marking completion as the final step means the next run ignores a half-finished one. In the refined version the manifest publish is that step. It’s separate because object storage gives you no atomic commit across many files.

A new source is a new file

The directory scan exists for this. Part 2 made the same claim from the test side, that a new source doesn’t require rewriting old tests. Adding a source means adding a spec file like the one above to the scanned directory, through a pull request, and nothing in ADF or the engine changes.

State derived from the target, and what it costs

The control-database design is stateful by construction. Watermark, last run and resume point live in tables that have to be updated transactionally for the next run to be correct, and a failed deployment or a half-finished run leaves them wrong, and somebody has to fix them by hand. The thread assignment stayed broken for the same reason: nobody could see what it was set to.

The thin design keeps its only record in the target. So there’s no state to corrupt, no table to rebuild after a bad deploy, and the engine keeps nothing between executions. There is also no database that has to be up and unlocked before a load can start, which caused a good part of the failed nights at the client. The deletion registry and the manifest are both written by the run into the target, and neither counts as control-table state. When a slice is wrong or doubtful, the repair is the forced full run from the command line: run full under a new run id, mark it complete, and the next delta derives from that. In the old design you had fresh data the next day if you were lucky. The full ELT ran in under an hour, so a break found in the run record at eight got fixed, rerun with a forced full load and reported on before ten, when the business needed the numbers.

Deriving state from the target gets harder when the source’s keys or timestamps don’t behave the way the load pattern assumes. That’s the objection I’d expect from anyone who has run a watermark load against a real source. Part 6 already covered hard deletes, since a watermark load cannot detect them and the registry carries the deletion. A key that doesn’t advance monotonically, or records that change retroactively, means the last completed run’s bounds no longer describe what the source will hand over next. That costs a wider re-read window overlapping the last slice, a snapshot pattern in place of a watermark, or accepting late corrections as a separate load.

Then there’s the third option. Managed ingestion tools such as Fivetran, Airbyte, Fabric Copy job and Lakeflow Connect keep their configuration and their watermark state inside the tool. All four now have a Terraform provider or an API for the configuration, so that half goes under version control. The watermark state is harder, since each of them keeps it inside the tool and exposes it partially at best. I haven’t run any of them in production. You’ll have to make that comparison yourself. Part 7 named Fivetran in History Mode and Airbyte with a CDC connector as the strongest starting points for managed SaaS API sources, and that stands.

Two things that release independently

A spec change is a pull request and a pipeline deploy. An engine change is a new wheel version. Neither requires the other, and a source can move to a new engine version, or stay on an old one, without its spec changing. The first design’s tables were in git too, or about half of them were, as a Visual Studio database project. Nothing deployed that project automatically, so every table change went into an environment by hand.

The contract and the delivery agreement in one file

Part 10 defined the contract as a bilateral agreement between producer and consumers with a schema section and a quality section, and Part 4B and Part 11 put freshness inside that quality section. Part 11 enforced it in front of Bronze and left open who runs the checks. They run in the engine at load time.

The delivery agreement is the source-facing half of that same agreement: what you get, when it’s available, and who on the source side is responsible when it isn’t there. It adds two things the quality section doesn’t include, an availability window and a source owner, which sit in the spec file next to the schema as available_by and owner above. The source owner is responsible for the source arriving inside its window. The layer owner from Parts 4B and 11 is a different person, the warehouse-side one who gets paged when a check trips.

With the window and the owner in the spec, the run record can say “source not available inside its window, owner crm-platform-team” instead of just a failure status. That separates a platform fault from a source fault. We didn’t get to implement this at the client. It belongs in the spec file, and it would have told us whose problem a missing load was before anyone opened a ticket. A missed window ends up as a recorded fact, the rows land on the next run, and that would usually be the end of it.

Three faces of one architecture

In Part 15’s terms, these objects are three things at once. The spec file per source is the parameterization and the engine that reads it is the framework, while the run record the engine writes back, with the run id and timestamp it stamps on every Bronze row, is the processing metadata and the pipeline-run half of monitoring. They’re one architecture because they’re the same few objects, each filling one of Part 15’s three roles. The run metadata is written by the load itself, so nobody maintains it. Part 5 made the same claim for the build side, where the catalog updates on merge.

The spec supplies the description, the owner, the grain and the classification for the landed source, and the run records supply freshness and source-to-landing lineage. Column descriptions for Silver and Gold still ship with the dbt model as Part 15 described. The human part, in Part 5’s terms, is authoring the spec.

What you can build from here

You now have four objects and one lifecycle, with a reason given for each. Metadata describes the differences between sources. Code describes how those differences are executed. The first design put the differences in tables that were dropped every night and the execution in procedures nobody could read, and most of that year went into working around it. The next post covers the load patterns the spec file can name and what each one costs, and the one after that covers monitoring.