Companion to the Data Engineering pillar, part 1 of 2. Two posts already sit under this pillar: Fabric + Mesh on AWS walks the technical foundation, and Behaviour-first governance in practice walks how governance artefacts get generated from platform events. This pair picks up the last piece of the same story — the pipelines that actually produce those events. Part 1 covers what a pipeline reads. Part 2 covers what it emits.
What the pillar and its companions leave
The pillar makes the argument for the Fabric + Mesh Hybrid at the level of what to build and why. The AWS reference makes it real at the level of accounts, catalogs, and grants. The governance companion makes it real at the level of how RoPA, RACI, and audit reports get generated from platform events.
The question that keeps coming back from readers of those posts is honest and correct: "but where is the actual data engineering?" You have accounts, LF-tags, grants, subscriptions, event streams. What produces those events? What moves data from a source system into a domain's producer account? What transforms it? What tests it before publish?
That is the pipeline half. These two posts walk it — end-to-end on the same AWS-native stack the earlier posts established, with the same insistence on metadata-first as the central operating principle. The principle itself comes from Data Fabric — the half of the Hybrid that treats active metadata as the platform's runtime driver, not a build-time artefact.
The central claim, worth stating once out loud:
Metadata-first pipelines beat retrofit metadata every time — and the retrofit is what kills the data platform in year two.
A note on what follows. The argument is illustrated with real artefacts rather than diagrams alone — YAML descriptors, a Python executor, a Great Expectations suite. The argument itself is architectural rather than language-specific: the same shape works anywhere you have a schema validator and a dispatch table, and the prose carries the thread for anyone who would rather skim the code blocks than read them line by line.
§ 1 — What "metadata-first" actually means
The phrase gets abused. "We have OpenLineage installed" is not metadata-first. "We emit some events" is not metadata-first.
It also gets read as a one-directional story — metadata as exhaust, something the pipeline gives off on its way past. That is half of it. Metadata crosses the pipeline boundary in both directions, and the inbound half is the one that decides whether you write forty pipelines or one.
Diagram read-out. The executor sits on the data path — source system in, data product out — but the diagram deliberately shows only what moves around that path. On the left, five declared inputs the job reads at run time instead of carrying inside its own source: schemas from the Glue Data Catalog, the job descriptor from the owning domain's repository, the target contract fixing output schema and SLA and LF-tags, the expectation suite holding the data-quality rules, and the policy metadata those tags resolve against. One generic executor interprets all of them — the box is singular on purpose: many descriptors, one implementation. On the right, three streams every run emits: the lineage event, the assertion results, and the provenance facet. The middle pair is the clearest illustration of the whole argument — the expectation suite goes in, its result comes out — and the same is quietly true of the rest: the descriptor naming source, target, and tags on the way in is most of the lineage event on the way out. All three outbound streams land in the governance layer, where RoPA, RACI, and audit reports become projections over them rather than documents somebody maintains by hand.
Inbound — the metadata a pipeline reads
Five kinds, and none of them belong inside the job's source code:
- Source schemas — what is arriving, with types, nullability, and a version. Runtime source of truth is the Glue Data Catalog for tables at rest, or the Glue Schema Registry where the source is streaming — two different services, and § 4 covers why the distinction changes where drift gets caught. The job reads the schema; it does not carry a hardcoded column list. Treated in § 4.
- Job descriptors — which source, which transformation steps with which parameters, which target. Declarative, versioned in Git, reviewed like code because it is code's input. Treated in § 2 and § 3.
- Target contracts — the output schema, the freshness SLA, and the LF-tags the resulting data product must carry. One declaration drives two things: what the job writes, and what gets registered in the catalog at publish. Treated in § 4.
- Expectation suites — the data-quality rules the output is asserted against: distributional bounds, referential integrity, freshness windows, accepted-value sets. Versioned next to the descriptor, owned by the domain that owns the product, read at run time. This is the entry that pairs most directly across the boundary — the rules go in, the assertion result comes out. Treated in § 5.
- Policy metadata — the LF-tag ontology and sensitivity classifications (Fabric + Mesh on AWS § 3), read from the catalog at run time. A reclassification upstream then propagates without anyone editing a pipeline. Treated in § 6.
The claim underneath this is the same one we made about admin screens in One skeleton, many screens, transplanted to the data tier: a pipeline is a descriptor, not a program.
The failure pair is identical too. Hand-write each pipeline and you get forty near-identical DAGs — directed acyclic graphs, the files an orchestrator like Airflow executes to run a pipeline — differing only in a table name. Fix a bug and you fix it in all of them, missing several. React against that with one large pipeline carrying a configuration flag for every case it has ever met, and you have moved the problem rather than solved it: now nobody can change the executor without breaking a domain three accounts away. The move that works is the same one that worked for screens — change what a pipeline is. Declare source, steps, target, and policy as data; keep one small generic executor that interprets the declaration.
The test for whether the boundary landed in the right place is equally concrete. Migrating an existing hand-written job onto the descriptor should produce a byte-identical output table and an identical lineage event. Zero diff, or the abstraction is wrong.
Outbound — the metadata a pipeline emits
The stricter test on this side: every pipeline job emits, as first-class outputs alongside its data, three metadata streams that downstream governance can consume without asking — lineage, assertion results, and provenance.
Part 2 treats all three in depth: OpenLineage as the vendor-neutral shape, DataZone as the AWS-native sink, the data-quality facet, and the provenance facet that audit reports are generated from. What matters here is only the direction. A pipeline that emits all three as first-class outputs generates the evidence the governance layer needs for free. A pipeline that has to be scanned, guessed at, or reverse-engineered later — by a discovery tool crawling query logs — turns that evidence into standing work: a reconstruction effort that has to be repeated every time the platform drifts, and is never quite finished.
One distinction is worth fixing before part 2 elaborates on it, because mixing up the two sides is the single most common mistake in quality programmes: the rules are inbound metadata, the results are outbound. An expectation suite is something the pipeline reads. An assertion result is something it produces. A quality programme that only has the first has a policy nobody can prove was applied.
The retrofit shape
The alternative is what most platforms end up with, and it fails in both directions. Outbound, a discovery tool scans CloudTrail, guesses at lineage from query patterns, and produces a graph that is partly wrong and permanently behind — without telling you which parts. Inbound, the column list lives inline in every job, so a source schema change becomes a repo-wide grep — and that is the good outcome, because at least those jobs failed. The bad one is the job that keeps running against a column that no longer means what it did, writing plausible-looking wrong values downstream for however long it takes somebody to notice. By then the remedy is not a code change but a backfill, and a backfill only works while the raw data still exists. Once the landing zone's lifecycle rules have expired the affected window, the corrupted output is simply what you have, and the honest next step is disclosure rather than repair. It is not that the tools are bad. It is that the truth was never declared anywhere a machine could read it, so everything downstream is reconstructing a shape only the job's original author ever knew.
Metadata-first fixes both by moving the source of truth out of the code — into a descriptor the pipeline reads, and an event stream the pipeline emits.
§ 2 — Anatomy of a job descriptor
Here is a real transform pipeline, declared. This is the entire artefact a domain engineer writes to add a data product — no DAG file, no job class, no orchestrator boilerplate:
# domains/fraud/pipelines/transactions_scored.yml
apiVersion: cordata.tech/v1
kind: TransformPipeline
metadata:
name: transactions-scored-daily
domain: fraud
owner: [email protected]
source:
kind: glue_table
database: fraud_raw
table: transactions
schema_version: 7
steps:
- id: filter_settled
kind: sql
query_file: sql/filter_settled.sql
- id: score
kind: sql
query_file: sql/score.sql
params:
model_version: "2026-07-fraud-v3"
target:
kind: iceberg
database: fraud_curated
table: transactions_scored
partition_by: [scored_date]
location: "s3://cordata-fraud-curated-eu-central-1/transactions_scored/"
contract:
freshness_sla: PT4H
lf_tags:
sensitivity: high
residency: eu
subject_type: customer
expectations:
suite: expectations/transactions_scored.yml
on_failure: block_publish
processing: # flows into the lineage job facet — part 2
purpose: fraud-detection
legal_basis: legitimate-interestSix blocks, each answering exactly one question: where the data comes from, what to do to it, where it goes, what must be true of it, what rules it is asserted against, and under what legal basis it is being processed at all.
Two details carry more weight than they look:
The only executable thing in the descriptor is SQL, referenced by path. sql/score.sql holds the domain's business logic, and it should — that is the part that genuinely differs between pipelines and genuinely belongs to the domain. Everything else is declaration. The line between the two is the line between what the domain owns and what the platform owns.
The processing block is compliance metadata living next to the code it describes. purpose and legal_basis are DSGVO Art. 30 fields. They are declared here, travel outward on the lineage event, and land in the generated processing record — which is why the Behaviour-first governance companion can project a RoPA without a compliance-team spreadsheet. The pipeline author declares the legal basis once, in the file they are already editing, and the record maintains itself.
The same schema, a different pipeline
A descriptor format that only fits the pipeline it was designed around is just a config file. Here is an ingestion pipeline in a different domain — different source technology, different output format, different failure policy — expressed in the identical schema:
# domains/policy/pipelines/claims_ingest.yml
apiVersion: cordata.tech/v1
kind: IngestionPipeline
metadata:
name: claims-ingest
domain: policy
owner: [email protected]
source:
kind: dms_landing # ← different reader
database: policy_raw
table: claims
schema_version: 3
steps:
- id: dedupe_cdc # ← CDC dedupe, not scoring
kind: sql
query_file: sql/dedupe_cdc.sql
target:
kind: parquet # ← different writer
database: policy_raw
table: claims_deduped
partition_by: [ingest_date]
location: "s3://cordata-policy-raw-eu-central-1/claims_deduped/"
contract:
freshness_sla: PT1H
lf_tags:
sensitivity: high
residency: eu
subject_type: customer
expectations:
suite: expectations/claims_deduped.yml
on_failure: quarantine # ← quarantine, not block
processing:
purpose: claims-processing
legal_basis: contract # ← different Art. 6 basisNothing in the executor changes to run this. Four fields differ in value — source.kind, target.kind, expectations.on_failure, processing.legal_basis — and each of those is a key into a registry or an enum the platform already implements. That is what the abstraction buys: the second pipeline costs a YAML file, and so does the fortieth.
How to define the descriptor schema
The schema itself is the design decision that everything else inherits from. Six rules produce one that survives contact with real domains:
- Declare intent, not mechanism.
kind: icebergsays what the output is.spark.sql.sources.partitionOverwriteMode: dynamicsays how one engine achieves it, and pins you to that engine forever. Every mechanism detail you admit into the descriptor is a migration you will pay for later. - Reference code by path; never inline it.
query_file: sql/score.sqlkeeps SQL in a.sqlfile where an IDE can lint it, a reviewer can read it in a diff, and a formatter can touch it. YAML block scalars holding SQL are how descriptors rot into unreadable programs. - Anything that varies per pipeline goes in the descriptor; anything constant goes in the executor. This is the whole test, and it is worth applying literally. If a field has the same value in every descriptor, it is executor configuration wearing a descriptor's clothes — delete it.
- No field may require its author to know how the executor is implemented. A domain engineer writes
on_failure: quarantinewithout knowing where the quarantine bucket is or which IAM role writes to it. The moment a field leaks implementation, the domain is coupled to the platform's internals and the ownership split in § 7 collapses. - Version the schema from the first descriptor.
apiVersion: cordata.tech/v1costs one line on day one and is what lets a futurev2land beside v1 rather than replacing it — § 3'sload()uses it to choose the model to parse against, so migrating the domains never becomes a flag day. - Validate in CI against the same model the executor uses at run time. One Pydantic model, imported by both. Two implementations of the same rules is two implementations that drift, and the CI one always drifts toward permissive.
The rule doing the most work is (3). It is also the one teams break first, usually by adding a field for a genuine one-off. When that happens, the honest move is the escape hatch in § 3 — not a new field that every descriptor will carry and one will use.
On the spelling, since it is borrowed and worth attributing: apiVersion group plus kind is Kubernetes' convention, chosen here because every platform engineer already parses it without being told, and because the group prefix keeps descriptors originating from different sources from colliding. The deviation is deliberate. Kubernetes nests the body under spec: because a controller writes status: back into the object; nothing writes back to a descriptor, so a spec: wrapper would be an indentation level implying a status that never arrives. And kind is the field doing least work today — both kinds currently validate against the same model. It is there to keep a directory of descriptors scannable at a glance, and to reserve the discriminator slot for when the two shapes genuinely diverge, which they will the first time an ingestion pipeline needs a CDC mode that a transform has no use for.
§ 3 — The generic executor
The descriptor above is inert without something to interpret it. That something contains no table names, no column lists, no SQL, and no domain logic of any kind — everything that differs between pipelines already lives in the descriptor. What is left is dispatch, and dispatch does not grow when a domain adds a product.
Start with the shape a descriptor has to satisfy. This is platform-owned, versioned, and validated in CI:
# platform/pipeline_runtime/descriptor.py (Python 3.12+)
from datetime import timedelta
from enum import Enum
from typing import Literal, Self
from pydantic import BaseModel, ConfigDict, Field, model_validator
class Strict(BaseModel):
"""Shared config for every descriptor model.
`frozen` stops anything mutating a descriptor after load.
`str_strip_whitespace` absorbs the trailing spaces hand-edited YAML
collects — without it `fraud_raw ` is a different database from `fraud_raw`.
"""
# Deliberately not strict=True: strict mode wants a timedelta instance and
# would reject `freshness_sla: PT4H`.
model_config = ConfigDict(extra="forbid", frozen=True, str_strip_whitespace=True)
class OnFailure(str, Enum):
BLOCK_PUBLISH = "block_publish"
QUARANTINE = "quarantine"
WARN = "warn"
class Metadata(Strict):
name: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
domain: str
owner: str # routed to on failure
class TableRef(Strict):
"""A catalog address. Source and Target are each one, plus their own extras."""
database: str
table: str
@property
def fqn(self) -> str:
return f"{self.database}.{self.table}"
class Source(TableRef):
kind: str # key into READERS
schema_version: int = Field(ge=1) # the catalog version this was written against
class Step(Strict):
id: str
kind: str # key into STEPS
query_file: str | None = None
module: str | None = None # the escape hatch — see below
params: dict[str, str] = Field(default_factory=dict)
@model_validator(mode="after")
def exactly_one_body(self) -> Self:
if bool(self.query_file) == bool(self.module):
raise ValueError(f"step {self.id}: set exactly one of query_file / module")
return self
class Target(TableRef):
kind: str # key into WRITERS
partition_by: list[str] = Field(default_factory=list)
location: str
class Contract(Strict):
freshness_sla: timedelta # ISO-8601, e.g. PT4H
lf_tags: dict[str, str] # validated against the ontology — § 6
class Expectations(Strict):
suite: str
on_failure: OnFailure
class Processing(Strict):
"""DSGVO Art. 30 fields. Travel outward on the lineage event."""
purpose: str
legal_basis: Literal[
"consent", "contract", "legal-obligation",
"vital-interests", "public-task", "legitimate-interest",
]
class Descriptor(Strict):
apiVersion: Literal["cordata.tech/v1"]
kind: Literal["TransformPipeline", "IngestionPipeline"]
metadata: Metadata
source: Source
steps: list[Step] = Field(min_length=1)
target: Target
contract: Contract
expectations: Expectations
processing: ProcessingTableRef is worth a word first, because the boundary it draws is easy to get wrong. A source and a target are both a catalog address — a database and a table — so that much is genuinely one concept and factoring it out gives both a shared fqn the readers, the run log, and part 2's lineage payload all need. What deliberately stays unfactored is kind: it appears on both, spelled identically and typed identically, but on a source it keys into READERS and on a target into WRITERS. Same name, different registries, so hoisting it would be shared spelling rather than shared meaning — the kind of tidy-looking inheritance that makes a schema harder to read later.
Three further choices are worth pulling out, because they are what makes the schema hold up rather than merely parse.
extra="forbid" turns a mistyped key into a failed pull request. The default — silently ignoring unknown fields — is how a descriptor ends up carrying on_fail: block for a year while the pipeline quietly runs with the default policy. legal_basis is a Literal of the six Art. 6 lawful bases rather than a free string, so an invented basis cannot reach the generated processing record. And the exactly_one_body validator is the kind of rule that has to live in the model rather than in the executor: catching query_file and module set together at PR time costs a reviewer nothing, while catching it at run time costs a failed nightly load.
Then the executor itself. Each numbered comment marks an inbound metadata kind from § 1, in order — and the branch in step 4 is where all three on_failure policies get their meaning:
# platform/pipeline_runtime/run.py
from collections.abc import Callable
from pathlib import Path
# The three registry contracts, stated once. Anything registered below has to
# satisfy one of them, which is what keeps "add a source type" a one-line change.
type Reader = Callable[[Source, Schema], DataFrame]
type Writer = Callable[[DataFrame, Target, dict[str, str]], None]
type StepRunner = Callable[[DataFrame, Step, Path], DataFrame]
READERS: dict[str, Reader] = {
"glue_table": read_glue_table,
"dms_landing": read_dms_landing,
"jdbc": read_jdbc,
"s3_parquet": read_s3_parquet,
}
WRITERS: dict[str, Writer] = {"iceberg": write_iceberg, "parquet": write_parquet}
STEPS: dict[str, StepRunner] = {"sql": run_sql, "python": run_python_module}
from .errors import PipelineError, UnknownApiVersion
# apiVersion is a dispatch key, not decoration. A future v2 model is registered
# beside v1 rather than replacing it, so descriptors written against v1 keep
# parsing while domains migrate one at a time.
MODELS: dict[str, type[Descriptor]] = {"cordata.tech/v1": Descriptor}
def load(path: Path) -> Descriptor:
raw = yaml.safe_load(path.read_text())
if (model := MODELS.get(raw.get("apiVersion"))) is None:
raise UnknownApiVersion(raw.get("apiVersion"), known=sorted(MODELS))
return model.model_validate(raw)
def run(descriptor_path: Path, run_id: str) -> None:
"""Entry point. Guarantees exactly one terminal event per run."""
pipeline = load(descriptor_path) # UnknownApiVersion escapes — see below
try:
execute(pipeline, descriptor_path.parents[1], run_id)
except PipelineError as exc:
emit(pipeline, run_id, exc.result, status="FAIL", error=exc) # part 2
raise
def execute(pipeline: Descriptor, domain_root: Path, run_id: str) -> None:
# Paths inside a descriptor (`sql/…`, `expectations/…`) are relative to the
# domain root, not to the descriptor — descriptors sit in <domain>/pipelines/.
# 1. source schema — resolved from the catalog, never hardcoded (§ 4)
schema = catalog.resolve(pipeline.source)
frame = READERS[pipeline.source.kind](pipeline.source, schema)
# 2. the declared steps — the domain's SQL, in order
for step in pipeline.steps:
frame = STEPS[step.kind](frame, step, domain_root)
# 3. policy metadata — tag values resolved against the ontology (§ 6)
tags = policy.resolve(pipeline.contract.lf_tags)
# 4. the declared expectation suite — rules in, result out (§ 5)
result = validate(frame, load_suite(domain_root / pipeline.expectations.suite))
if not result.success:
match pipeline.expectations.on_failure:
case OnFailure.BLOCK_PUBLISH:
raise PublishBlocked(pipeline.metadata.name, result)
case OnFailure.QUARANTINE:
WRITERS[pipeline.target.kind](frame, quarantined(pipeline.target), tags)
emit(pipeline, run_id, result, status="QUARANTINED") # part 2
return
case OnFailure.WARN:
log.warning("%s: %s", pipeline.metadata.name, result.summary)
# 5. write, carrying the contract
WRITERS[pipeline.target.kind](frame, pipeline.target, tags)
emit(pipeline, run_id, result, status="COMPLETE") # part 2One property of that shape is worth stating as a rule, because it is the difference between a platform that can answer for itself and one that cannot: every run emits exactly one terminal event. Success, quarantine, or failure — the governance layer hears about all three.
Getting that from discipline alone does not work. Three hand-written emit() calls at three exits is three chances to forget, and the fourth exit somebody adds next year will forget. So it comes from structure instead: execute raises, run catches at a single boundary, and every executor failure inherits from PipelineError so that one except covers all of them. A pipeline that dies on schema drift therefore still produces a FAIL event carrying the reason — rather than leaving a silent gap in the lineage graph where a run should be, which is exactly the shape a supervisor cannot distinguish from "nobody scheduled it".
The hierarchy that makes the single except possible is small enough to show in full, and lives in its own module so that catalog.py and policy.py can raise without importing run.py, which already imports them:
# platform/pipeline_runtime/errors.py
class PipelineError(Exception):
"""Base for every failure the executor raises, so one boundary catches all."""
result: Validation | None = None # most failures have no assertion result to report
class UnknownApiVersion(PipelineError): ...
class SchemaDrift(PipelineError): ...
class UnknownTagKey(PipelineError): ...
class UnknownTagValue(PipelineError): ...
class PublishBlocked(PipelineError):
def __init__(self, name: str, result: Validation) -> None:
super().__init__(f"{name}: expectations failed before publish")
self.result = result # the one failure that carries an assertion resultOnly PublishBlocked has a result to hand the terminal event, because it is the only failure that got as far as running the suite. The rest inherit the None default, which is why emit() accepts Validation | None — a run that died on schema drift still produces an event, it just has an error to report rather than an assertion result.
The honest edge: UnknownApiVersion is raised inside load(), before there is a validated descriptor to attribute an event to. Nothing useful can be emitted for a file the executor cannot parse — that failure belongs to CI, which validated the descriptor at pull-request time precisely so it never reaches a scheduler.
That is the whole implementation, modulo the reader and writer bodies. Every descriptor across every domain runs through these forty lines.
The registries are where the executor grows, and the contract they impose is deliberately narrow — a reader takes a Source and a resolved Schema, and returns a frame that conforms to that schema or raises:
# platform/pipeline_runtime/readers.py
def read_glue_table(source: Source, schema: Schema) -> DataFrame:
df = spark.read.format("iceberg").load(source.fqn)
return schema.enforce(df) # project, cast, fail on mismatch
def read_dms_landing(source: Source, schema: Schema) -> DataFrame:
# DMS CDC lands one file per batch with Op/before/after columns;
# collapse to the latest row per key before anyone downstream sees it.
raw = spark.read.format("parquet").load(dms_prefix(source))
df = collapse_cdc(raw, keys=schema.primary_keys)
return schema.enforce(df)Both readers end at schema.enforce, and that is the contract: every reader returns a frame matching the pinned schema, so nothing downstream — no step, no suite, no writer — has to defend against a source-shaped surprise. The messiness of change-data-capture from AWS DMS — the Database Migration Service, which replicates a source database into S3 one batch of inserts, updates and deletes at a time — stays inside twelve lines, where collapse_cdc is a plain testable function rather than a block copy-pasted into every job that reads from DMS.
Worth separating that from the dedupe_cdc step the claims-ingest descriptor in § 2 declares, since the two sound like the same job and are not. The reader collapses the change log to one row per key — DMS mechanics, true of every source that arrives this way, and therefore the platform's. The step drops the domain's notion of a duplicate, such as the same claim submitted twice under two ids, which only the policy team can define. The ownership line in § 7 runs exactly between them.
Adding a source technology is one entry in READERS plus a function honouring that contract — not a new pipeline, and not a new flag on an existing one.
Those registries look like configuration, so why are they literals? Because their values are functions, and configuration can only hold a string naming one. Moving them out means import paths in YAML — a plugin system wearing configuration's clothes, converting an import error at module load into a failed lookup mid-run inside one domain account. It buys nothing either, since the set of readers changes only when somebody writes a reader, which ships code anyway. Design rule 3 again, pointed at the executor: which readers exist is a property of the executor version, not of any pipeline.
The exception is a domain with a proprietary source nobody else will read, where a platform pull request is the wrong coupling; Python entry points are the mechanism, at exactly that deferred-error cost. One honest gap: the python step kind below covers custom transforms, not custom sources, because it runs on a frame already read. An exotic source is still a platform change today.
A run against the transform descriptor from § 2 reads like this:
$ python -m pipeline_runtime domains/fraud/pipelines/transactions_scored.yml
[9f3c] descriptor transactions-scored-daily (fraud) apiVersion v1 OK
[9f3c] schema fraud_raw.transactions pinned v7, catalog v7 OK
[9f3c] read glue_table 4,182,905 rows
[9f3c] step filter_settled (sql) 1,143,207 rows
[9f3c] step score (sql) 1,143,207 rows
[9f3c] expect transactions_scored 11 passed, 0 failed
[9f3c] policy sensitivity=high residency=eu resolved against ontology
[9f3c] write fraud_curated.transactions_scored iceberg, 4 partitions
[9f3c] emit COMPLETE → lineage adapter inputs=1 outputs=1
Every line corresponds to one of the five inbound metadata kinds being resolved, or to the outbound event that resolution makes possible. Nothing in that trace is specific to fraud scoring. Point the same command at domains/policy/pipelines/claims_ingest.yml and the shape is identical — different reader, different writer, quarantine instead of block_publish, and a contract legal basis on the emitted event.
The failure path is the more instructive one. When the upstream team adds a column to fraud_raw.transactions:
[a17e] descriptor transactions-scored-daily (fraud) apiVersion v1 OK
[a17e] schema fraud_raw.transactions pinned v7, catalog v8
[a17e] FAILED SchemaDrift: + merchant_category_code (string, nullable)
descriptor pins v7 — bump the pin to accept
Nothing was read, nothing was written, nothing was published, and the error names the exact column and the exact remedy. § 4 covers why that is the correct behaviour rather than an inconvenience.
Where the generic executor stops working
Every abstraction has a boundary, and pretending this one does not is how you arrive at that same over-configured pipeline by a different route.
Three shapes genuinely resist declaration: transformations whose logic cannot be expressed as ordered SQL steps (iterative algorithms, graph traversals, anything needing a real control flow); sources that do not behave like tables, such as a vendor API needing its own pagination and retry handling; and one-off backfills that will run twice and never again.
The escape hatch is the python step kind, pointing at a domain-owned module:
steps:
- id: reconcile
kind: python
module: domains.fraud.transforms.reconcile_late_settlements
params:
lookback_days: "30"The point of routing the escape hatch through the descriptor rather than around it: the module is arbitrary Python, but the pipeline still declares its source, its target, its contract, its expectation suite, and its legal basis — so lineage, assertions, and tag propagation all still happen. The custom logic is confined to the one place it is genuinely needed, and the metadata contract survives intact. An escape hatch that bypasses the executor entirely would forfeit all of that — which is exactly what a hand-written DAG does.
When not to build this at all
Below roughly twenty pipelines, the executor costs more than it saves. The descriptor schema, the CI validation, the reader and writer registries, the documentation a domain engineer needs to write their first descriptor — that is a platform investment, and it pays back on repetition. Writing a pipeline by hand is not the problem. Writing the fortieth one by hand is.
§ 4 — Schemas and contracts, resolved from the catalog
schema_version: 7 is the most consequential line in the descriptor, and the whole of its weight is in the number.
The executor resolves that reference at run time against the Glue Data Catalog:
# platform/pipeline_runtime/catalog.py
from .descriptor import Source
def resolve(source: Source) -> Schema:
current = glue.get_table(DatabaseName=source.database, Name=source.table)
version = int(current["Table"]["VersionId"])
if version != source.schema_version:
raise SchemaDrift(
f"{source.fqn} is at v{version}, descriptor pins "
f"v{source.schema_version}. "
f"Diff: {diff_columns(source, version)}"
)
return Schema.from_glue(current)resolve takes the whole Source rather than a reference string, which is why there is no parsing here at all. An earlier draft of this schema addressed the table with a URI — catalog://fraud_raw/transactions@v7 — and that turned out to be a small self-inflicted wound: the descriptor already declares database and table, so the URI restated both and left two places to keep in step. Design rule 3 caught it. The version is the only thing the reference was really carrying, so the version is all that remains.
The interesting behaviour is the failure. When an upstream team adds a column, the source table moves to v8, the descriptor still pins v7, and the run fails loudly with a column-level diff — before writing anything. The domain engineer then reads the diff, decides whether the new column matters, and bumps the pin in a pull request. That PR is the record of the decision.
Compare with the two common alternatives. SELECT * silently absorbs the new column and passes it downstream, where a consumer's schema-on-read breaks a week later with no obvious cause. A hardcoded column list silently ignores it, which is fine until the upstream team renames something and the pipeline keeps running against a column that no longer means what it did. Neither produces a moment where a person decides anything.
The target side works the same way in reverse. The contract block is what the pipeline promises its subscribers — output schema, freshness SLA, and the LF-tags the product carries. It drives what the writer emits and what gets registered in the catalog at publish, from one declaration. Subscribers reading the catalog are reading the same contract the pipeline was executed against, not a wiki page describing it.
Data Catalog, not Schema Registry — and why that limits the design
Worth stating precisely, because AWS ships two services with "Glue" and "schema" in the name and they solve different problems.
The Glue Data Catalog is a metastore for data at rest: databases, tables, columns, partitions, and a linear version history of each table definition. It is what Athena, EMR, Redshift Spectrum, LakeFormation, and the Glue jobs in these posts all read, and it is what a descriptor's database, table, and schema_version address above. The Glue Schema Registry is a separate service for data in motion — Avro, JSON Schema, and Protobuf message schemas for MSK, Kinesis, and Flink — and its distinguishing feature is that it enforces compatibility rules at registration.
That difference is not cosmetic, and it exposes the honest limitation of everything above. A registry rejects an incompatible schema when the producer tries to register it, before a single message ships. The Data Catalog has no equivalent moment to hook: a table definition changes whenever the upstream team runs DDL or a crawler completes, and the first party in a position to notice is whoever reads it next. Run-time SchemaDrift is therefore not the ideal place to catch drift — it is the earliest place available for tables at rest.
Streaming sources should push that check left into the registry rather than rediscovering it per consumer, which is one of the concrete reasons the streaming shape needs its own treatment instead of a paragraph here. A descriptor for a streaming source would name a compatibility mode rather than pin a catalog version — a different mechanism for the same intent.
§ 5 — Expectation suites as declared input
Assertions are the one inbound stream with a genuine tool choice attached, so it is worth saying why this architecture reaches for Great Expectations rather than the more obvious option.
The obvious option, if the transform tier is dbt, is dbt's own tests — not_null, unique, accepted_values, relationships declared in the same YAML as the models, extended by dbt-expectations where the built-ins run out. For a shop that is dbt end to end, that is the right answer and adding a second assertion framework is a cost with no return.
The deeper limit is not which pipelines run dbt, though — it is which ETL stage dbt can see. dbt is the T. It can reach a little further than that — sources: with tests is a real feature, so freshly landed tables can be asserted too — but it checks what has already arrived in a warehouse it can query, on its own invocation. The E and the L are therefore covered after the fact, if at all, and a truncated source file or a half-applied CDC batch has already landed by the time anything runs against it.
The same gap then repeats per pipeline. A descriptor's steps dispatch through the STEPS registry — SQL today, a Python module for the exceptions in § 3, whatever a domain registers next — and the claims-ingest descriptor in § 2 never touches dbt at any point. The rule the platform actually needs is that every data product is asserted before publish, whatever engine produced it and at whatever stage, and no tool that sees one engine and one stage can enforce it.
Great Expectations buys exactly that reach: it validates a frame, not a warehouse, so the same suite shape works behind any reader, at any step, in any pipeline. Its results also arrive as structured objects rather than log lines, which is what lets § 3's validate hand a uniform assertion result to emit() regardless of what ran. The cost is honest — a second framework to learn, and statistical expectations that need a stored baseline to compare against.
So it is better understood here as an inbound metadata stream than as a testing tool: a versioned artefact the pipeline reads, whose assertion results it emits.
The suite lives in the owning domain's repository, alongside the descriptor that references it and the SQL those steps run. Every path inside a descriptor — sql/…, expectations/… — resolves against that domain root:
domains/fraud/
├── pipelines/transactions_scored.yml # the descriptor
├── sql/filter_settled.sql # step bodies
├── sql/score.sql
└── expectations/transactions_scored.yml # the suite below
# domains/fraud/expectations/transactions_scored.yml
suite_name: transactions_scored
expectations:
- type: expect_column_values_to_not_be_null
kwargs: {column: tx_id}
- type: expect_column_values_to_be_unique
kwargs: {column: tx_id}
- type: expect_column_values_to_match_regex
kwargs: {column: iban, regex: '^[A-Z]{2}[0-9]{2}[A-Z0-9]{1,30}$'}
- type: expect_column_kl_divergence_to_be_less_than
kwargs: {column: fraud_score, partition_object: baseline_fraud_score, threshold: 0.15}
- type: expect_table_row_count_to_be_between
kwargs: {min_value: 850_000, max_value: 1_400_000}The first three are hygiene, the sort of check a type system would give you for free. The last two are different in kind. Distributional drift on the fraud score — KL divergence against a stored baseline — catches a model regression that every schema test in the world would pass. Row-count boundaries catch upstream deletion bugs before they cascade into a consumer's dashboard.
partition_object: baseline_fraud_score names that baseline rather than inlining it. The executor resolves the name against a baselines/ directory in the same domain repository, which keeps a hundred bin weights out of a file people have to read — and means re-baselining arrives as a diff with an author on it, rather than as a number quietly changing inside a suite. "The model drifted" and "we decided the new distribution is fine" must not look the same in the history.
Placement matters as much as content. The suite runs at the pipeline boundary — after the declared steps, before the write. Not scattered through the transformation. Not on a scheduled scan hours later. At the exact seam where the pipeline is about to say "here is the output I promise my subscribers", the promise is checked. That is why on_failure: block_publish is a meaningful setting: at that seam, failing means nothing is published, and no consumer ever sees the bad batch.
The three on_failure modes exist because not every rule deserves the same response. block_publish for correctness invariants — a null primary key should stop the world. quarantine for volume anomalies, where writing to a side location and paging the descriptor's owner beats both publishing and halting. warn for rules still being tuned, which is the honest setting for a new distributional bound whose threshold nobody trusts yet. A suite where everything blocks gets disabled within a month.
Ownership follows the same line as the descriptor: the domain owns the suite, because the domain is the only party that knows what fraud_score is supposed to look like. Platform owns the runner. Changing a rule is a pull request against the domain's repository — which means the quality policy has a commit history, an author, and a reviewer, rather than being a paragraph in a governance document.
What the suite produces — the per-run assertion result, and what governance does with it — is part 2.
§ 6 — Policy metadata and LF-tag resolution
The descriptor names tag values:
contract:
lf_tags:
sensitivity: high
residency: eu
subject_type: customerIt does not define what high means, what it permits, or who may read data carrying it. That vocabulary — the LF-tag ontology from Fabric + Mesh on AWS — lives in the governance account and is owned by the governance function. The executor resolves against it at run time:
# platform/pipeline_runtime/policy.py
def resolve(declared: dict[str, str]) -> dict[str, str]:
ontology = lakeformation.list_lf_tags() # governance-owned vocabulary
for key, value in declared.items():
if key not in ontology:
raise UnknownTagKey(key)
if value not in ontology[key]:
raise UnknownTagValue(key, value, allowed=ontology[key])
return declaredTwo properties fall out of this:
Reclassification propagates without touching pipelines. When governance decides that residency: eu now implies a narrower grant expression, the change happens once, in the ontology. Every pipeline tagging its output residency: eu inherits the new meaning on its next run. This is the difference between policy-as-code and policy-as-a-column-you-copied.
Unknown values fail before they reach production. The same validation runs in CI against the descriptor, so a typo like sensitivity: hgh fails at pull-request time with the allowed set in the error message — not against a table that has already been written with a tag nothing grants against. Validating the same rule in both places, from the same ontology, is what makes the CI check trustworthy rather than a second implementation that drifts.
§ 7 — Ownership, storage, and the standard that does not exist
Ownership splits three ways, and it maps exactly onto the three-way contract from the Governance pillar:
- Platform owns the executor and the descriptor schema — the shape any descriptor has to satisfy.
- Domain owns the descriptor instances for its own data products, the SQL those descriptors reference, and its expectation suites.
- Governance owns the policy vocabulary those descriptors reference — the LF-tag ontology, the sensitivity levels, the legal-basis values.
That split is the reason the model holds under a federated operating model rather than quietly re-centralising. The platform team never reviews a domain's business logic; the domain never edits the executor; governance never touches a pipeline. Each side changes only the artefacts it owns, and the descriptor schema is the contract between them.
Storage follows from ownership, and the thing that trips teams up is that a descriptor has three homes, and they are easy to mix up:
- Source of truth — the owning domain's Git repository. Not an AWS account. The descriptor, the SQL its steps reference, and the expectation suite sit together under that domain's CODEOWNERS and are reviewed like any other change. The signed commit is then already the change-authorisation record that Behaviour-first governance in practice § 4 describes — that evidence costs nothing extra to produce.
- Run-time copy — a config prefix in the producer account. The executor runs where the data is, so the descriptor has to be reachable from there; the domain's CI syncs it on merge. This copy is derived and disposable. Nobody edits it in place, and rebuilding it from the repository is always safe.
- The schema it must satisfy — a versioned package the platform publishes. Imported by the domain's CI and by the executor, which is what makes the pull-request check trustworthy rather than a second implementation that drifts.
Schemas themselves live in the Glue Data Catalog as the runtime source of truth, per § 4.
The tempting shortcut is a single central repository holding every domain's descriptors, administered by the platform team. That reproduces precisely the bottleneck this architecture exists to remove: every pipeline change becomes an item in a queue the platform team owns. Descriptors are domain artefacts and belong in domain-owned repositories — whether that means one repository per domain or a monorepo with per-directory CODEOWNERS matters far less than who approves the merge.
Where the code runs follows the same logic. The executor runs inside the producer account, next to the data, for exactly the reason the DMS instance does: source credentials and raw rows never cross the account boundary. What is central is the executor artefact — platform builds and publishes a versioned image, each domain account consumes it. One implementation, many accounts. Orchestration is the one piece that genuinely wants to sit in the middle, and part 2 works through why, including where that stops being true.
Put end to end, defining and onboarding a new pipeline looks like this:
Diagram read-out. A domain engineer opens a pull request containing three files and no code — descriptor, SQL, expectation suite. CI validates it against the platform's descriptor schema and checks that the pinned schema_version is still the current one in the catalog, so a typo or a stale version pin is caught by a reviewer rather than by a pager. On merge the signed commit becomes the change-authorisation record, and CI syncs the descriptor into the producer account's config prefix. Note what does not happen at that point: no new code is built or deployed, because the platform-published executor is already there — onboarding a pipeline is a configuration change, not a release. At run time the executor reads the source schema from the catalog, runs the declared steps, asserts the suite, resolves the policy tags, and emits its three streams to the governance account, which is where part 2 picks up.
One honest asymmetry: the inbound side has nothing with OpenLineage's consolidation behind it. Parts of a descriptor are covered by open standards — dbt's source and model YAML is the de-facto shape for the transform tier, and where dbt is the engine you should adopt it rather than invent a parallel one.
What none of them yet does is span source, steps, target, contract, expectations and policy in a single declaration — and that is the whole reason apiVersion in § 2 reads cordata.tech/v1 rather than naming something a standards body owns. A small in-house descriptor, published as a versioned JSON Schema, is the pragmatic answer until one of those partial standards grows to cover the rest.
The consolation is that the descriptor is your schema, in your repository, and migrating it later is a mechanical transformation over a directory of YAML files. That is a considerably better position than migrating forty hand-written DAGs to anything at all.
Part 2 — the evidence a pipeline emits — covers OpenLineage as the vendor-neutral lineage backbone, the AWS-native adoption playbook via DataZone's PostLineageEvent, assertion results as a data-quality facet, and the provenance facet that audit reports are generated from.
The code in this post runs. cordata-tech/pipeline-runtime is the executor above with DuckDB and local Parquet standing in for Glue and Spark. A clean clone runs both descriptors from § 2 end to end with no cloud account, and reproduces the SchemaDrift failure from § 4 from a fixture. Building it also turned up five places where these two posts were wrong or would not run as printed — those are written up in the repository, and the substantive ones have since been corrected here.
If you have built a descriptor-driven pipeline platform — or built one and found the abstraction boundary in the wrong place — I would genuinely like to hear where it bent. The door is open at cordata.tech/contact. Related reading: the Data Engineering pillar for the composed argument, the Fabric + Mesh on AWS reference for the account-per-domain topology this pipeline sits on, and Behaviour-first governance in practice for how the governance layer reads what these pipelines produce.