Companion to the Data Engineering pillar, part 2 of 2. Part 1 — a pipeline is a descriptor, not a program covered the metadata a pipeline reads: source schemas, job descriptors, target contracts, expectation suites, policy metadata, and the one generic executor that interprets them. This part covers what it emits — and why that is the entire governance story.
Recap — where part 1 left off
Part 1 ended with a descriptor-driven executor writing a data product and calling emit(...) on each of its three exits — complete, quarantined, and failed — deferring the definition to here. block_publish is not a fourth exit: it raises, and that raise lands on the same single boundary catching schema drift and unknown policy tags, so a blocked publish leaves through failed carrying the assertion result that blocked it. § 4 defines emit() in full, once the three streams it has to carry are established.
The three streams it emits were named in part 1 § 1, and the distinction that matters most is worth restating once, because the two sides are easy to mix up: the rules are inbound metadata, the results are outbound. An expectation suite is something a pipeline reads. An assertion result is something it produces. A programme holding only the first has a policy nobody can prove was applied.
Three outbound streams, then:
- Lineage — inputs consumed, outputs produced, transformation identity. § 1 and § 2.
- Assertion results — what the declared expectation suite reported for this run. § 3.
- Provenance — what code, what commit, what configuration produced this run. § 4.
There is a structural payoff from part 1 worth flagging up front: because the pipeline was declared rather than written, most of what these events need is already sitting in the descriptor: the source and target, the legal basis, the pinned schema version. Emission is largely a matter of forwarding what was already declared — which is why metadata-first pipelines get their governance evidence close to free, and retrofit pipelines never quite do.
A note on what follows, as in part 1. The argument is illustrated with real artefacts rather than diagrams alone — OpenLineage payloads, emitter configuration, a small Python adapter, Terraform. It is an architectural argument rather than a language-specific one, and the prose carries the thread for anyone who would rather skim the payloads than read them field by field. The JSON is worth slowing down for in one place only: the two facets in § 1 that the governance layer reads directly.
§ 1 — OpenLineage as the vendor-neutral backbone
OpenLineage is an open specification for lineage events, governed under the LF AI & Data Foundation. Worth being explicit about why it is the choice here, because "the industry consolidated on it" is the weakest of the available reasons: lineage is the one artefact in this stack that has to outlive the tools that produced it, so the only format worth writing years of history into is one that no single runtime, cloud, or catalog vendor can change unilaterally. The argument holds on a shop running one engine with no plans to move: the neutrality is insurance bought before it is needed, not a reaction to already having four engines.
Its structure is deliberately small: a RunEvent (one per job execution) carries an input list, an output list, a job identifier, a run identifier, and facets — extensible metadata attached to any of those.
{
"eventType": "COMPLETE",
"eventTime": "2026-08-04T09:14:22.123Z",
"producer": "https://github.com/cordata-tech/pipeline-runtime/tree/2.4.1",
"schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent",
"run": {
"runId": "b8c1c9e3-5a1e-4c0f-9a1a-2c9f8b1a5e12",
"facets": {
"processing_engine": {"name": "spark", "version": "3.5.1"},
"nominalTime": {"nominalStartTime": "2026-08-04T09:00:00Z"}
}
},
"job": {
"namespace": "cordata.fraud",
"name": "transactions-scored-daily",
"facets": {
"sourceCodeLocation": {"type": "git", "path": "sql/score.sql",
"url": "https://github.com/cordata-tech/domains-fraud"},
"processing": {"purpose": "fraud-detection", "legal_basis": "legitimate-interest"}
}
},
"inputs": [
{"namespace": "cordata.fraud", "name": "fraud_raw.transactions"}
],
"outputs": [
{"namespace": "cordata.fraud", "name": "fraud_curated.transactions_scored",
"facets": {"schema": {"fields": [{"name": "tx_id", "type": "STRING"},
{"name": "fraud_score", "type": "DOUBLE"}]}}}
]
}One thing to note before the fields that matter. producer and schemaURL are required on every event and are shown here in full; every facet additionally requires its own _producer and _schemaURL, which are omitted throughout this post so the fields under discussion stay readable. A payload copied from here will be rejected by a validating consumer until those are added — the spec is the reference, not these excerpts. The event above is also a merged view rather than one emitter's output: the runtime integration contributes the mechanical facets and the executor contributes the declared ones, both keyed to the same runId. § 4 unpacks that.
Two facets in the example above are worth naming because the governance layer reads them directly:
job.facets.processing— thepurposeandlegal_basisfields flow directly into the DSGVO Art. 30 record projection in the Behaviour-first governance companion § 2. The RoPA is generated from these fields. That means every job's lineage event carries the compliance metadata for its own processing activity — not a separate compliance-team spreadsheet. Note where those two values came from: theprocessingblock of the descriptor in part 1 § 2. The pipeline author declared them once, in the file they were already editing.output.facets.schema— the schema of what the job produced. Downstream consumers subscribing via DataZone see this schema in the catalog automatically. No manual catalog registration step. Note who fills it in: the runtime integration, not the executor. Theopenlineage-sparkplugin reads the column list off the Spark execution plan;emit()never sets it, because a descriptor declares which table to write, not which columns the transformation will produce. § 4 works through how two emitters end up contributing to one event.
Worth being clear about what that second facet is not, because the instinct is to expect a pointer: it embeds the field list rather than referencing a registry entry, and that is the right way round for evidence. A reference to "v7" is only as good as the promise that v7's definition never changes; an embedded snapshot records the shape this run actually produced and stays true whatever the registry does afterwards. The registry reference is a separate concern with a facet of its own — CatalogDatasetFacet carries the framework, catalog name, and a metadataUri, which is where a Glue Data Catalog or Glue Schema Registry pointer belongs. Part 1's inbound side pins a version; the outbound side records an observation. Same split as everywhere else in these two posts, one layer down.
The ecosystem is what makes acting on that argument cheap. Every non-trivial pipeline runtime has an integration — Spark (openlineage-spark), Airflow (built into core since 2.7), dbt (the dbt-ol wrapper), Trino, Flink, Dagster. Instrumenting a job is typically one config line plus a package install, not a code rewrite.
The alternative — proprietary lineage from a single vendor's SaaS catalog — makes two properties of the lineage graph contingent on a commercial relationship: its portability and its coverage. When the catalog vendor changes, the lineage becomes a data-migration project. When a new pipeline runtime enters the stack (say, adding Flink alongside existing Spark), the proprietary catalog only sees half the graph.
§ 2 — The AWS-native adoption playbook: DataZone as the OpenLineage sink
The natural question at this point on AWS-native shops is: "do we need to run a separate lineage backend like Marquez?"
Since 2024 the answer has been no. Amazon DataZone accepts OpenLineage events directly via its PostLineageEvent API — announced as preview in July 2024, since matured. The events land in DataZone's own lineage store, and the events + LF-tag metadata + subscription log all live in the same catalog the governance layer reads from.
The pattern:
Diagram read-out. Every pipeline runtime emits OpenLineage events at its natural boundary — Spark on Glue via the openlineage-spark plugin, Airflow on MWAA via the built-in OpenLineage provider, dbt via the dbt-ol wrapper, and the executor itself via emit(). All emitters post to a thin adapter Lambda (or Kinesis stream, depending on volume), which wraps the payload unchanged in a PostLineageEvent request. The lineage lands in DataZone's catalog, alongside the assets, subscriptions, and LF-tag policies the Fabric + Mesh on AWS reference established.
The emitter integrations
Glue Spark jobs — the openlineage-spark plugin. Attach the plugin as a Glue job argument; it emits events for every dataframe operation without touching the job code. One Glue-specific trap is worth showing rather than describing: job parameters are a key/value map, so --conf can only appear once. Every additional Spark setting has to be chained inside that single value, --conf and all:
# Glue job parameters — two keys, not five
--extra-jars s3://cordata-artifacts/openlineage-spark_2.12-1.34.0.jar
--conf spark.extraListeners=io.openlineage.spark.agent.OpenLineageSparkListener
--conf spark.openlineage.transport.type=http
--conf spark.openlineage.transport.url=https://lineage-adapter.cordata.internal
--conf spark.openlineage.namespace=cordata.fraud
Note the artifact name: the jar is Scala-version-suffixed (_2.12), and it has to match the Scala version of the Glue runtime you are on.
MWAA Airflow — the built-in OpenLineage provider. Get the package right here, because the older one still turns up in search results: OpenLineage moved into Airflow core under AIP-53 with Airflow 2.7. The standalone openlineage-airflow is deprecated and must not be installed on Airflow 2.8 or later — on MWAA, that means the provider, always:
# requirements.txt
apache-airflow-providers-openlineage==2.20.0
# MWAA "Airflow configuration options" — each becomes AIRFLOW__OPENLINEAGE__*
openlineage.transport {"type": "http", "url": "https://lineage-adapter.cordata.internal"}
openlineage.namespace cordata.orchestration
dbt — the dbt-ol wrapper. This one is not a plugin, and the difference matters. OpenLineage support is not a dbt Labs feature — dbt-core #11750 is still an open request — so the integration is openlineage-dbt, a wrapper maintained by the OpenLineage project that you substitute for the dbt binary. It runs dbt untouched, then reads manifest.json and run_results.json once the run finishes and emits from those artefacts. Configuration is environment variables, not profiles.yml:
pip install openlineage-dbt
OPENLINEAGE__TRANSPORT__TYPE=http
OPENLINEAGE__TRANSPORT__URL=https://lineage-adapter.cordata.internal
OPENLINEAGE_NAMESPACE=cordata.transform
# then substitute the wrapper for the binary
dbt-ol run --select tag:curated
Being a post-processor rather than an in-process listener has one consequence worth knowing before you rely on it: events arrive after the run completes, not as it progresses, so a dbt process killed before it writes its artefacts emits nothing at all. That is the opposite of the guarantee part 1's run() boundary provides, and it is worth knowing which of your runtimes give you a terminal event under failure and which quietly give you silence.
The executor itself — emit(). The fourth emitter in the diagram is the pipeline runtime from part 1, posting to the same adapter as the plugins. It is there because nothing else in the stack can produce two of the three streams: no plugin knows the descriptor's legal_basis, and none can hash the declaration that produced the run. § 4 defines it.
Note — Great Expectations can also emit on its own. Worth knowing before deciding not to use it. openlineage-integration-common[great-expectations] ships an OpenLineageValidationAction that goes in a checkpoint's action list:
{"name": "open_lineage",
"action": {"class_name": "OpenLineageValidationAction",
"module_name": "openlineage.common.provider.great_expectations",
"openlineage_host": os.getenv("OPENLINEAGE_URL"),
"openlineage_namespace": os.getenv("OPENLINEAGE_NAMESPACE")}}The package pins great-expectations>=1.0.0, while the published example still shows the older checkpoint style — so check the wiring against the Great Expectations version you are actually on.
The executor attaches the result itself instead, for one reason: it already holds the Validation, having run the suite at the pipeline boundary. Attaching it there puts the assertion result on the same terminal event as the lineage and the provenance, keyed to the run that produced all three, rather than depending on a separately configured checkpoint action landing a matching event of its own. Use the action when Great Expectations runs somewhere the executor does not — a scheduled scan over a warehouse table, say. Inside a descriptor-driven pipeline the suite is already part of the run, so the result travels with it.
The thin adapter — enveloping, not translation
The adapter is thinner than "translation" suggests, because DataZone does not want a different payload. PostLineageEvent takes the OpenLineage RunEvent as the raw request body — the API reference is explicit that "only open-lineage run event are supported as events" — and carries DataZone's own identifiers elsewhere in the HTTP request. Nothing inside the payload is rewritten.
Diagram read-out. One event in, one HTTP request out, split three ways. The DataZone domain identifier addresses the request through the URI path; a hash of the payload becomes the Client-Token header that makes retries idempotent; and the OpenLineage event itself is the body, byte for byte as the emitter produced it. That last property is the reason this stays a thin adapter rather than a mapping layer with its own release cycle — and it is why an OpenLineage upgrade that adds a facet needs no change here at all.
# adapter.py — simplified core
import hashlib
import json
import os
import boto3
datazone = boto3.client("datazone")
DOMAIN_ID = os.environ["DATAZONE_DOMAIN_ID"] # dzd-… , one per DataZone domain
def handle(openlineage_event: dict) -> None:
payload = json.dumps(openlineage_event, sort_keys=True)
datazone.post_lineage_event(
domainIdentifier=DOMAIN_ID,
# Idempotency is per event, not per run. One runId legitimately
# produces several events — a START, the plugin's COMPLETE, and our
# own — so keying the token on runId alone would drop all but the
# first and quietly undo the accumulation § 4 depends on. Hashing the
# payload dedupes genuine retries and nothing else. sort_keys makes
# that hash stable across dict orderings.
clientToken=hashlib.sha256(payload.encode()).hexdigest(),
event=payload,
)Two constraints are worth knowing before this carries production traffic. The event body is capped at 300 KB — a wide schema facet, or column-level lineage over a hundred-column table, gets closer to that than it sounds, and the adapter is the right place to notice and fail loudly rather than let events disappear into a ValidationException. And clientToken is capped at 128 characters with pattern [\x21-\x7E]+, which a 64-character hex digest satisfies with room to spare.
That token stops being theoretical as soon as volume moves you from API Gateway to Kinesis: delivery is at-least-once, and Lambda retries a failed batch whole, so a batch failing on its four-hundredth event redelivers the first three hundred and ninety-nine. The sort_keys=True is what keeps the digest stable across those retries, since Python serialises dicts in insertion order.
When you do not need any of this. openlineage-python ships an AmazonDataZoneTransport that calls the same PostLineageEvent API from inside the client, so a single Python producer can point at DataZone by configuration and stop reading here. The adapter earns its place in the topology this post is about, where the producers are Spark, dbt and Airflow rather than one Python process, and a transport living inside the Python client cannot serve them. Worth knowing before you adopt it, though: that transport sends no clientToken at all, so the idempotency above is something you get from the adapter and not from the library.
§ 3 — Assertion results as a data-quality facet
OpenLineage says what happened. The assertion result says whether the data that came out was any good — and it is the stream that turns data quality from a stated intention into evidence a regulator can read.
Part 1 § 5 covered the expectation suite as declared input: a versioned artefact, owned by the domain, read at the pipeline boundary after transformation and before publish. This is the assertion result it produces, and § 4's emit() puts it on the same run:
{
"eventType": "OTHER",
"run": {"runId": "b8c1c9e3-5a1e-4c0f-9a1a-2c9f8b1a5e12"},
"job": {"namespace": "cordata.fraud", "name": "transactions-scored-daily.validate"},
"inputs": [{
"namespace": "cordata://fraud",
"name": "fraud_curated.transactions_scored",
"inputFacets": {
"dataQualityAssertions": {
"assertions": [
{"assertion": "expect_column_values_to_not_be_null", "column": "tx_id", "success": true},
{"assertion": "expect_column_kl_divergence_to_be_less_than", "column": "fraud_score",
"success": false, "expected": "0.15", "actual": "0.19"}
]
}
}
}],
"outputs": []
}Two things about where that sits look wrong at first, and both are the spec rather than a preference. The assertion result rides on an input, and the dataset it names is the one this pipeline just produced. dataQualityAssertions is typed as an InputDatasetFacet, so inputFacets is its only standard home — and the dataset an assertion refers to is an input to the assertion, whatever it was to the pipeline. Both reference integrations agree: dbt's emits it against a <model>.test job, and Great Expectations' own OpenLineage action emits inputs=[…], outputs=[]. Put it on the output dataset's facets instead and the event is still well-formed — facet maps are open — but the result sits where nothing looks for it, which is the quiet kind of wrong. Second, the type is OTHER rather than a second COMPLETE, because part 1 § 3 guarantees exactly one terminal event per run, and OTHER is what the spec reserves for additional metadata accumulating against a run id — the same mechanism § 4 depends on.
One run id, then, carrying both the lineage and the assertion results for the same batch. That they arrive as two events costs a consumer nothing, since joining on the run id is what a lineage store is for, and the co-location is what makes the next move possible.
Who reads this downstream. Not one signal for one audience — it lands in the same catalog three roles already work in, and each asks a different question of it.
- The producing domain. The descriptor names an
owner, and a failing suite is that team's problem first. Part 1'son_failurealready decided whether this batch publishes; what emitting the result adds is history — the same expectation failing twice in a week is a different problem from failing once. - The governance or data-quality function. Results accumulate into a per-dataset series with the run, the batch and the expectation attached. A quality policy breach becomes something with evidence beside it rather than a claim somebody has to reconstruct afterwards.
- Consumers of the dataset. The facet rides on the dataset's node in the catalog, so a team deciding whether to subscribe — or a dashboard about to refresh against it — can see whether the most recent batch passed before depending on it. That is the question a consumer actually has, and usually the one nobody can answer.
Automated enforcement sits at the far end of that range: a rule like "suspend a subscription's grant when a producer's daily assertions fail three times in a rolling week" only becomes expressible once the results are events, and it is the same operating pattern as Behaviour-first governance in practice, where the enforcement and the evidence are the same act. Worth being clear that it is an escalation rather than a default, though — it answers a quality question by withdrawing access, and those are not the same axis.
Note what the failing assertion in the payload above is: expect_column_kl_divergence_to_be_less_than on fraud_score, actual 0.19 against a declared bound of 0.15. No schema test would have caught it — the column is present, correctly typed, and non-null. The model's output distribution simply moved. The value of emitting the result rather than merely running the check is that this drift is now a queryable event with a timestamp, attached to the exact batch that exhibited it, in the same store the auditor already has access to.
The difference between what a schema test can catch and what this one caught also decides who can write the check. A schema expectation — column present, correctly typed, non-null — needs only the shape of the data, so a platform or pipeline-execution team can write it, or generate it from the catalog entry without knowing what the table is for. A bound of 0.15 on the KL divergence of fraud_score needs someone who knows what that score means, what its distribution normally looks like, and what a shifted one implies about the model behind it. No platform team has that, and no central quality service acquires it by being given more tables. That is why part 1 § 5 puts the suite in the owning domain's repository: shape checks can be centralised, semantic ones only exist where the meaning does.
§ 4 — Provenance: the run's identity
Lineage says which datasets were involved. Provenance says what recipe produced this specific batch — the code, the commit, the configuration, the engine: not the run's data; the run's identity.
This is the stream that answers the question a BaFin or DORA conversation eventually arrives at: which code computed this number, against which inputs, and who authorised it?
In a descriptor-driven platform, provenance is unusually cheap to produce, because the descriptor from part 1 already is most of the recipe. Hash it and you have captured the source, the steps, the target, the contract, and the legal basis in a single field:
"run": {
"facets": {
"processing_engine": {"name": "spark", "version": "3.5.1"},
"cordata_provenance": {
"descriptor_path": "domains/fraud/pipelines/transactions_scored.yml",
"descriptor_sha256": "9c1f4b7e…",
"source_table": "fraud_raw.transactions",
"source_schema_version": 7,
"descriptor_git_commit": "4af3958a…",
"descriptor_git_commit_signed": true,
"executor_version": "pipeline-runtime 2.4.1",
"step_params": {"model_version": "2026-07-fraud-v3"}
}
}
}Four of those fields do the real work.
descriptor_sha256 pins the exact declaration this run executed — not the descriptor as it looks today, but as it looked when this run executed. source_schema_version pins what the input was shaped like, which is the difference between "this number came from transactions" and "this number came from transactions as it was defined before the merchant-category column landed". step_params captures the values that were substituted into the declared steps, which is where model_version: 2026-07-fraud-v3 lives — the single most audit-relevant fact about a scoring run.
Two identities are being recorded here and it is worth not conflating them. descriptor_git_commit is the domain's repository — the commit that carried this descriptor, which is what makes the reproduction below possible. executor_version is the platform's, and the two move independently: a domain can change a threshold without the runtime changing at all, and the runtime can ship a release without any descriptor moving. "Which code computed this number" needs both answers, which is why both are on the event.
A fair objection at this point: what stops the recorded hash being changed after the fact? On its own, nothing — a number sitting in an event proves very little, and an auditor is right to treat it as a claim rather than a fact. What makes it evidence is that it is reproducible. descriptor_git_commit sits in the same facet, Git objects are content-addressed, so anyone can run git show <commit>:<descriptor_path> | sha256sum a year later and check that three independent artefacts agree: the hash recorded at run time, the descriptor as that commit stored it, and the signature over that commit. Disagreement between them is itself the finding. Forging the set means rewriting Git history, altering the lineage event, and re-signing under a trusted key — and the event is not somewhere the producing team can reach: PostLineageEvent targets a DataZone domainIdentifier, and the topology § 1 puts the DataZone domain in the central governance account, not in the producer accounts where the pipelines run.
descriptor_git_commit_signed is the one that closes the loop. The commit that introduced this descriptor version was signed, reviewed, and merged under the GitOps flow the Behaviour-first governance companion § 5 describes. So the provenance facet does not merely record what ran — it chains to who authorised it to run, through an artefact nobody can retroactively edit. A run whose provenance points at an unsigned commit is itself a finding.
The retrofit version of this is a deployment log — which model version went live when — updated by hand, by whoever remembers.
One run, three streams — the emit() part 1 kept deferring
Part 1's executor calls emit(...) on all three of its exits and defers the definition to here. It is the smallest interesting function in the system, because by this point every field it needs has already been declared or computed:
# platform/pipeline_runtime/emit.py
from dataclasses import replace
from openlineage.client import OpenLineageClient
from openlineage.client.event_v2 import Job, Run, RunEvent, RunState, set_producer
from openlineage.client.facet_v2 import error_message_run
# Becomes the producer on every event; the client supplies schemaURL.
set_producer(f"https://github.com/cordata-tech/pipeline-runtime/tree/{__version__}")
# Resolves its transport from the environment, so nothing here names a
# destination — where events go is deployment configuration, not a pipeline's.
client = OpenLineageClient()
STATE = {
"COMPLETE": RunState.COMPLETE,
"FAIL": RunState.FAIL,
"QUARANTINED": RunState.FAIL, # a quarantined run did not publish
}
def emit(pipeline: Descriptor, run_id: str, result: Validation | None,
status: str, error: PipelineError | None = None) -> None:
# 1. Run facets — provenance always, plus the reason if this run died (§ 4)
run_facets = {"cordata_provenance": provenance_facet(pipeline)}
if error is not None:
run_facets["errorMessage"] = error_message_run.ErrorMessageRunFacet(
message=str(error), programmingLanguage="python")
# 2. Datasets — what the run actually touched. A failed run touched nothing;
# a quarantined one wrote to quarantined(target), not the declared one.
match status:
case "FAIL":
inputs, outputs = [], []
case "QUARANTINED":
inputs, outputs = [dataset(pipeline.source)], [dataset(quarantined(pipeline.target))]
case _:
inputs, outputs = [dataset(pipeline.source)], [dataset(pipeline.target)]
job = Job( # job id + job facets
namespace=f"cordata.{pipeline.metadata.domain}",
name=pipeline.metadata.name,
facets={"processing": pipeline.processing.model_dump()}, # § 1
)
# 3. The terminal event — exactly one per run, carrying the five parts § 1 named
client.emit(RunEvent(
eventType=STATE[status],
eventTime=now_utc_iso(),
run=Run(runId=run_id, facets=run_facets), # run id + run facets
job=job,
inputs=inputs, # input list — consumed
outputs=outputs, # output list — produced
))
# 4. The assertion result, when there is one to report, against the dataset
# that was actually tested (§ 3). OTHER, not a second COMPLETE: part 1
# promises one terminal event per run, and this is not it.
if result is not None:
tested = quarantined(pipeline.target) if status == "QUARANTINED" else pipeline.target
client.emit(RunEvent(
eventType=RunState.OTHER,
eventTime=now_utc_iso(),
run=Run(runId=run_id), # same run — see § 4 on accumulation
job=replace(job, name=f"{job.name}.validate"),
inputs=[input_dataset(tested, inputFacets={
"dataQualityAssertions": result.as_facet()})],
outputs=[],
))Those are our own helper functions rather than calls into the client library, and all of them are mechanical. dataset() turns a descriptor's source or target block into an OpenLineage InputDataset or OutputDataset — a namespace, a name, and any facets passed in. input_dataset() is the same thing pinned to the input side, which the validation event needs because the dataset it asserts on is this pipeline's target. quarantined() derives the quarantine location from the declared target and comes from part 1, which uses it in the same branch. provenance_facet() assembles the eight fields above. replace is dataclasses.replace — the validation job differs from the pipeline's by its name alone. The client's transport is configured out of band, which is what lets the same emit() reach § 2's adapter in production and a file in a test. Descriptor, Validation and PipelineError are part 1's types.
Three facets, three sections of this post, one run. Note how little of it is computed: the namespace and job name come from the descriptor's metadata, the processing facet is the descriptor's processing block serialised as-is, the input and output datasets are its declared source and target. Only the run ID, the timestamp, and the assertion result are new information. That is the structural payoff flagged in the recap — emission is mostly forwarding what was already declared.
The match is the only real decision in the function, and it is there because the three exits genuinely touched different things. Worth not reading it as the three on_failure branches from part 1 § 3 in new clothing, though — those only run when the suite fails, and these are terminal states. QUARANTINED is the only one-to-one: COMPLETE covers a clean pass and a warn, while FAIL absorbs block_publish together with every failure that never reached the suite at all. One consequence is worth stating plainly, because it is the case where the split into two events earns itself: a blocked publish has no output dataset at all — nothing was written — so an assertion result attached to the output would have nowhere to go. Attached to the tested dataset it survives, and a FAIL event naming the error is followed by an OTHER event naming the expectation that caused it. A governance layer counting failures and one asking which rule fired are both answered from the run id, without either having to reach into the expectation store.
The event a failed run produces. Part 1 § 3 guarantees that every run emits exactly one terminal event, so a failure produces a payload rather than a gap — and it is the more detailed of the two, because it has to account for itself. When a run aborts on schema drift, run() catches at its boundary and emit() produces this:
{
"eventType": "FAIL",
"eventTime": "2026-08-05T03:02:11.880Z",
"producer": "https://github.com/cordata-tech/pipeline-runtime/tree/2.4.1",
"schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent",
"run": {
"runId": "a17e2b90-…",
"facets": {
"errorMessage": {
"message": "fraud_raw.transactions is at v8, descriptor pins v7. Diff: + merchant_category_code (string, nullable)",
"programmingLanguage": "python"
},
"cordata_provenance": {
"descriptor_sha256": "9c1f4b7e…",
"source_schema_version": 7,
"descriptor_git_commit": "4af3958a…"
}
}
},
"job": {"namespace": "cordata.fraud", "name": "transactions-scored-daily"},
"inputs": [], "outputs": []
}Empty inputs and outputs are not an omission — they are the fact being recorded. Nothing was read and nothing was written, and the lineage graph should say so rather than imply a partial run. The provenance facet still lands, because which descriptor version failed is exactly what the person fixing it needs, and the errorMessage carries the column-level diff that part 1 § 4's SchemaDrift assembled.
A missing event is the real problem: it looks exactly like a pipeline nobody scheduled. A FAIL event carrying its reason is what makes the run debuggable at all.
How this coexists with the plugin-based emission in § 2. Those two mechanisms look contradictory — the openlineage-spark plugin advertises that it emits without touching job code, and here is job code emitting. Both are running, and they are doing different work. The plugin derives the base event from the Spark plan: it sees inputs, outputs, column-level schema, row counts, and it sees them more accurately than any hand-written call would. What it categorically cannot know is the descriptor's legal_basis, the assertion results, or the hash of the declaration that produced the run — none of that exists in the Spark plan.
OpenLineage's data model resolves this cleanly, because events accumulate against a runId rather than replacing one another. So the plugin emits the mechanical truth, emit() sends the declared truth keyed to the same run, and the backend merges them into one lineage record. The alternative — disabling the plugin and constructing everything by hand — is available and occasionally right for non-Spark runtimes, but it means reimplementing column-level lineage extraction, which is precisely the work the plugin exists to do.
Three shapes rather than two, once dbt is in the stack: an in-process listener that watches execution as it happens (Spark, Airflow), a reconstruction after the fact from whatever the run left behind (dbt-ol reading dbt's artefacts once the run is over), and explicit calls like emit() contributing what only the descriptor knows. They compose because all three key to the same runId. But only the first and third can report a run that died — a post-processor needs the process to survive long enough to write its artefacts — which is the practical reason a dbt-heavy stack still wants an executor boundary of its own rather than trusting the wrapper for terminal events.
§ 5 — Orchestration: central scheduler, decentralised compute
The examples so far run Airflow on MWAA, and nothing in the argument depends on that. What the orchestrator has to do here is narrow: authenticate into a producer account on a schedule, trigger work, stay out of the way. Step Functions, Dagster, or anything that can assume a role at a fixed time will do it, and none of that changes what a pipeline emits — so the choice belongs to whoever has to operate the thing.
Note what the orchestrator is and is not doing in the descriptor-driven platform part 1 describes. There is still a DAG — but it holds no pipeline logic. It schedules execute(descriptor_path, run_id) and handles retries, backfills, and dependencies between products, while the source, the steps, the target, the contract and the policy all live in the descriptor. So every DAG ends up the same shape, and the file shrinks to a loop over the domain's descriptor directory — the second-largest reduction in hand-written code after the executor itself.
The question that does matter is one the account-per-domain topology forces: if every domain owns its own account, does every domain run its own orchestrator?
Almost certainly not, and the reason is the pricing model rather than the architecture. Decentralise what is priced per use; centralise what is priced by capacity. Glue is billed in DPU-seconds, so running transform jobs inside eight producer accounts costs the same as running them in one — decentralising compute is free, and it is also mandatory, because that is what keeps source credentials and raw rows inside the domain boundary. An always-on scheduler inverts that: you pay for the control plane whether or not anything runs that hour, and eight of them means eight to buy and eight to patch.
So the default shape is a single orchestrator in a shared automation account, assuming a narrow per-domain role to trigger work that executes inside the producer account. The orchestrator schedules and tracks; it never touches a row. That split also buys the thing per-domain orchestrators make genuinely hard — cross-domain dependencies, where the claims product cannot start until the policy product has landed. With one scheduler that is an edge in a graph; with eight it becomes an event bus between orchestrators, and a debugging surface nobody wanted. A regulated slice that cannot share a scheduler with the rest is worth the duplicated control plane, but as a named exception rather than the pattern.
The failure mode to avoid is a central orchestrator that grows domain logic. The moment it starts holding transformation SQL or per-domain branching, it stops being a scheduler and becomes the monolith the mesh was built to retire — and it will be owned by the platform team, which is how the central bottleneck comes back through the side door.
§ 6 — What a domain's infrastructure has to know
Very little, and that is the point. The Terraform standing a domain up needs almost nothing about the pipeline, because part 1's descriptor already holds it:
# live/prod/domains/fraud/terragrunt.hcl — transform jobs
transform_jobs = [
{
name = "transactions_scored_daily"
descriptor = "domains/fraud/pipelines/transactions_scored.yml"
openlineage_enabled = true # attaches the openlineage-spark plugin
},
]Two things about that entry. It names a descriptor, not a script path plus an input prefix plus an output prefix — those three moved into the file the domain owns, rather than sitting in Terraform where the platform team would own them. And OpenLineage is a flag on the infrastructure rather than a change to the job: the plugin attaches at the Glue job definition, so instrumenting a pipeline is not something a domain engineer has to remember.
The module behind those inputs creates a Glue job, because that is what this stack runs — and the entry is that short precisely because there is one engine. Run two, and something has to say which. Note where that something goes: an engine = "emr" beside descriptor in the Terraform entry, never a new field in the descriptor. The descriptor says what a pipeline reads and produces; the infrastructure says what runs it. Keeping the engine on that side of the line is what lets forty descriptors survive a migration off Glue without being touched. Ingestion is wired the same way, with the replication instance in the producer account for the reason § 5 gives about compute — a central one would concentrate every domain's raw rows in a single place.
Where the graph starts. OpenLineage has integrations for Spark, dbt, Flink, Hive and SQL; there is none for DMS. A replication task landing rows in raw/ emits no RunEvent, so the source system is not a node in the lineage graph.
That is a boundary rather than a hole, and a defensible one — the platform's responsibility begins at ingestion, the way an immigration desk's begins at the border. What arrives is recorded even though the journey before it is not. The task that landed those rows is declared in the Terraform above, with its source endpoint, target prefix and LF-tags version-controlled and PR-reviewed, so "where did raw/transactions/ come from" has an answer.
What is genuinely missing is narrower: you can say which task ingests that prefix, but not which replication run landed a particular batch. Per-table provenance, not per-batch. If that distinction matters — and for a regulated slice it might — the fix is a wrapper emitting a RunEvent around the replication task, which is an addition rather than a redesign.
§ 7 — The semantic layer above the data products
Above the data products a pipeline produces sits one more tier. The semantic layer — dbt's, or Cube, or Malloy — exposes typed metrics like monthly_recurring_revenue, policy_attach_rate and fraud_recall_at_30_days, with their business definitions versioned in Git, backed by dbt models, queried through a semantic API. BI tools, notebooks and LLM agents all reach for monthly_recurring_revenue and get the same number, defined the same way, backed by the same lineage.
That last consumer is why the tier stopped being optional. Raw SQL against a governed lake is unsafe for an agentic consumer; a typed semantic API above it is what makes an agent's question answerable without hallucination.
For the pipeline layer these two posts cover, the semantic layer is the consumer contract. Every transform pipeline producing a metric-underlying dataset publishes its schema through DataZone; the semantic layer references those datasets and defines metrics on top; consumers query the semantic layer rather than the raw tables. Metrics become discoverable, governed and consistent by construction, instead of re-derived per BI tool.
The Data Engineering pillar named this tier in "Where AI belongs in a Fabric + Mesh Hybrid", and the Fabric + Mesh on AWS reference § 4 called it "the tier DataZone does not give you". A dedicated post — The semantic layer is how AI reads your data safely — dbt Semantic Layer, Cube, and the DACH audit case — is in the backlog and will take it as its own subject. Worth separating which audit question belongs to which tier, because they are easy to swap. "Which code produced this number" is the pipeline layer's, and § 4's provenance facet answers it. The semantic layer's question is a different one: whether the monthly_recurring_revenue in a board pack and the one in a regulatory return are the same definition, who changed that definition, and when. A metric's meaning lives in no pipeline, which is why no amount of lineage answers it. For the pipeline half, the placement is what matters. Pipelines produce data products; the semantic layer is where those products become consumable across domains, because a metric defined once means the same thing in claims as it does in fraud — which is the federation problem a mesh creates and has to solve. Worth keeping that distinct from a governed business glossary, though: the glossary is the catalog's job, and DataZone has CreateGlossaryTerm for it. The catalog settles what a term means; the semantic layer makes every consumer compute it the same way.
§ 8 — What this does not cover
Three specific gaps these two posts deliberately leave for other treatments:
- Streaming pipelines. MSK Connect + Kinesis Firehose + Flink is a genuinely different shape — the concept of "run boundaries" for lineage becomes fuzzier, expectations run continuously rather than per-run, and the account topology has to hold up under always-on connectors. Deserves a separate post; the patterns above assume batch and scheduled runs.
- MLOps pipelines. Feature stores, model-training pipelines, model-serving lineage — different again, and the unit of lineage stops being a table. OpenLineage has no ML integration and no model or feature-store facets at all, so model lineage means defining custom ones; and the assertion story overlaps with model-quality metrics without being the same thing.
- The event-collection plumbing itself. The Behaviour-first governance companion § 1 takes these events as given — it names the sources its projections read from, then moves on to what can be built once they land. These two posts supply the pipeline half of that assumption, not the lineage-event-sink infrastructure half. The DataZone
PostLineageEventadoption playbook in § 2 covers the sink for AWS-native shops; for teams running Marquez or a homegrown sink, that specific plumbing is its own decision.
Closing the arc
Two pillars and four companion posts, all of them serving one sentence from the Data Governance pillar: the artefacts — policies, RACIs, RoPAs — are byproducts of that behaviour, not deliverables. Everything since has been an argument that this is buildable rather than aspirational, and each post carried one part of the load:
- The Data Engineering pillar made the architectural case — Data Fabric as the machinery, Data Mesh as the organisation, composed rather than chosen between.
- Fabric + Mesh on AWS made it concrete: account-per-domain, the LF-tag ontology, two-layer DataZone segmentation, in Terraform that runs.
- Behaviour-first governance in practice built the four projections on top — RoPA from lineage, RACI from catalog ownership, the audit report from the subscription log, change authorisation as an evidenced workflow — and named the three event sources they read from.
- Part 1 turned the pipeline into a declaration, so those events would have something honest to describe.
- This part made the declaration emit its own evidence.
Worth being precise about what that delivers against what was promised, because the promise was specific. Behaviour-first governance in practice § 1 names three event sources its projections read from: OpenLineage events, the DataZone subscription log, and CloudTrail data events. These two posts supply the first of those three, and only that one. The subscription log is DataZone's to produce; the per-query access log is LakeFormation's and CloudTrail's. What the pipeline layer owes the governance layer is the lineage stream — and each piece of it has a named destination:
job.facets.processing, carryingpurposeandlegal_basisstraight from the descriptor, is what the RoPA projection in that post's § 2 reads. The DSGVO Art. 30 record is generated per run, from a field the pipeline author wrote once.cordata_provenance, anddescriptor_git_commit_signedin particular, is what lets the change-authorisation workflow in Behaviour-first § 5 chain a published number back to the reviewed commit that authorised the code producing it.dataQualityAssertionsextends the same operating pattern to data quality, which the original four mechanisms did not cover: a grant that suspends on repeated assertion failures is a policy expressed as behaviour, and it needs a per-run result to act on.
Something is still maintained, of course — descriptors get edited, expectation suites get tuned, the executor ships releases. What stops being maintained is the evidence. The RoPA, the provenance trail, the quality history: nobody writes those, and nobody has to, because a pipeline running correctly leaves them behind. Define the descriptors well once and the artefacts keep their own pulse — which is the pillar's sentence, arrived at from the other end.
The code in this post runs. cordata-tech/pipeline-runtime emits the events above for real — spec-shaped OpenLineage, provenance facet included — to a file rather than an adapter, so a clean clone can read what it produced with no AWS account. Building it found the placement error in § 3 that this post used to have; that and four others are written up in the repository.
If you have shipped a metadata-first pipeline layer on AWS-native — or shipped one and found the retrofit shape irrecoverable — I would genuinely like to hear where the seams showed. The door is open at cordata.tech/contact. Related reading: part 1 — a pipeline is a descriptor, not a program for the inbound half; the Data Engineering pillar for the composed argument; the Fabric + Mesh on AWS reference for the account-per-domain topology these pipelines sit on; and Behaviour-first governance in practice for how the governance layer reads the events these pipelines emit. Two follow-ups still in the queue: The semantic layer is how AI reads your data safely — the dbt Semantic Layer / Cube / Malloy trade-off with the DACH audit case — and DSGVO Art. 17 across a mesh on the erasure mechanics when a customer's data lives in eight accounts.