Begleitbeitrag zum Data-Engineering-Grundpfeiler, Teil 1 von 2. Zwei Beiträge stehen bereits unter diesem Grundpfeiler: Fabric + Mesh auf AWS geht das technische Fundament durch, und Behaviour-first Governance in der Praxis zeigt, wie Governance-Artefakte aus Plattform-Ereignissen erzeugt werden. Dieses Paar greift das letzte Stück derselben Geschichte auf — die Pipelines, die jene Ereignisse überhaupt erst produzieren. Teil 1 behandelt, was eine Pipeline liest. Teil 2 behandelt, was sie emittiert.
Was der Grundpfeiler und seine Begleitbeiträge offen lassen
Der Grundpfeiler führt das Argument für den Fabric + Mesh Hybrid auf der Ebene was zu bauen ist und warum. Die AWS-Referenz macht es konkret auf der Ebene Konten, Kataloge und Grants. Der Governance-Begleitbeitrag macht es konkret auf der Ebene wie RoPA, RACI und Audit-Berichte aus Plattform-Ereignissen erzeugt werden.
Die Frage, die von Lesern dieser Beiträge immer wieder zurückkommt, ist ehrlich und berechtigt: „aber wo ist eigentlich das Data Engineering?“ Sie haben Konten, LF-Tags, Grants, Subscriptions, Ereignisströme. Was produziert diese Ereignisse? Was bewegt Daten aus einem Quellsystem in das Producer-Konto eines Fachbereichs? Was transformiert sie? Was prüft sie vor der Veröffentlichung?
Das ist die Pipeline-Hälfte. Diese beiden Beiträge gehen sie durch — Ende zu Ende auf demselben AWS-nativen Stack, den die früheren Beiträge etabliert haben, mit demselben Beharren auf Metadata-first als zentralem Betriebsprinzip. Das Prinzip selbst stammt aus Data Fabric — jener Hälfte des Hybrids, die aktive Metadaten als Laufzeit-Treiber der Plattform behandelt, nicht als Build-Zeit-Artefakt.
Der zentrale Anspruch, einmal laut ausgesprochen:
Metadata-first-Pipelines schlagen nachgerüstete Metadaten jedes Mal — und das Nachrüsten ist es, was die Datenplattform im zweiten Jahr umbringt.
Eine Anmerkung zum Folgenden. Das Argument wird mit echten Artefakten illustriert statt nur mit Diagrammen — YAML-Deskriptoren, ein Python-Executor, eine Great-Expectations-Suite. Das Argument selbst ist architektonisch und nicht sprachspezifisch: dieselbe Form funktioniert überall dort, wo es einen Schema-Validator und eine Dispatch-Tabelle gibt, und der Fließtext trägt den roten Faden auch für alle, die die Code-Blöcke lieber überfliegen als Zeile für Zeile lesen.
§ 1 — Was „Metadata-first“ tatsächlich bedeutet
Der Begriff wird missbraucht. „Wir haben OpenLineage installiert“ ist nicht Metadata-first. „Wir emittieren ein paar Ereignisse“ ist nicht Metadata-first.
Er wird außerdem als eindirektionale Geschichte gelesen — Metadaten als Abgas, etwas, das die Pipeline im Vorbeigehen abgibt. Das ist die halbe Miete. Metadaten überqueren die Pipeline-Grenze in beide Richtungen, und die eingehende Hälfte entscheidet darüber, ob Sie vierzig Pipelines schreiben oder eine.
Diagramm-Lesehilfe. Der Executor sitzt auf dem Datenpfad — Quellsystem hinein, Datenprodukt hinaus — aber das Diagramm zeigt bewusst nur, was sich um diesen Pfad herum bewegt. Links fünf deklarierte Eingaben, die der Job zur Laufzeit liest, statt sie im eigenen Quellcode zu tragen: Schemata aus dem Glue Data Catalog, der Job-Deskriptor aus dem Repository des besitzenden Fachbereichs, der Zielvertrag, der Ausgabeschema, SLA und LF-Tags festlegt, die Expectation Suite mit den Datenqualitätsregeln und die Policy-Metadaten, gegen die jene Tags aufgelöst werden. Ein generischer Executor interpretiert sie alle — die Box ist mit Absicht Singular: viele Deskriptoren, eine Implementierung. Rechts drei Ströme, die jeder Lauf emittiert: das Lineage-Ereignis, die Assertion-Ergebnisse und das Provenance-Facet. Das mittlere Paar illustriert das ganze Argument am deutlichsten — die Expectation Suite geht hinein, ihr Ergebnis kommt heraus — und dasselbe gilt still auch für den Rest: der Deskriptor, der Quelle, Ziel und Tags auf dem Weg hinein benennt, ist der Großteil des Lineage-Ereignisses auf dem Weg hinaus. Alle drei ausgehenden Ströme landen in der Governance-Schicht, wo RoPA, RACI und Audit-Berichte zu Projektionen darüber werden statt zu Dokumenten, die jemand von Hand pflegt.
Eingehend — die Metadaten, die eine Pipeline liest
Fünf Arten, und keine davon gehört in den Quellcode des Jobs:
- Quellschemata — was ankommt, mit Typen, Nullbarkeit und einer Version. Laufzeit-Quelle der Wahrheit ist der Glue Data Catalog für ruhende Tabellen oder die Glue Schema Registry, wenn die Quelle Streaming ist — zwei verschiedene Dienste, und § 4 erklärt, warum die Unterscheidung ändert, wo Drift gefangen wird. Der Job liest das Schema; er trägt keine fest verdrahtete Spaltenliste. Behandelt in § 4.
- Job-Deskriptoren — welche Quelle, welche Transformationsschritte mit welchen Parametern, welches Ziel. Deklarativ, in Git versioniert, wie Code reviewt, weil es Eingabe für Code ist. Behandelt in § 2 und § 3.
- Zielverträge — das Ausgabeschema, das Frische-SLA und die LF-Tags, die das entstehende Datenprodukt tragen muss. Eine Deklaration treibt zwei Dinge: was der Job schreibt und was beim Publish im Katalog registriert wird. Behandelt in § 4.
- Expectation Suites — die Datenqualitätsregeln, gegen die die Ausgabe geprüft wird: Verteilungsgrenzen, referenzielle Integrität, Frischefenster, erlaubte Wertemengen. Neben dem Deskriptor versioniert, im Besitz des Fachbereichs, dem das Produkt gehört, zur Laufzeit gelesen. Dies ist der Eintrag, der am direktesten über die Grenze hinweg gepaart ist — die Regeln gehen hinein, das Assertion-Ergebnis kommt heraus. Behandelt in § 5.
- Policy-Metadaten — die LF-Tag-Ontologie und Sensitivitäts-Klassifizierungen (Fabric + Mesh auf AWS § 3), zur Laufzeit aus dem Katalog gelesen. Eine Umklassifizierung stromaufwärts propagiert dann, ohne dass jemand eine Pipeline anfasst. Behandelt in § 6.
Der Anspruch darunter ist derselbe, den wir über Admin-Screens in Ein Gerüst, viele Screens gemacht haben, verpflanzt auf die Datenebene: eine Pipeline ist ein Deskriptor, kein Programm.
Das Fehlerpaar ist ebenfalls identisch. Schreiben Sie jede Pipeline von Hand, bekommen Sie vierzig nahezu identische DAGs — Directed Acyclic Graphs, die Dateien, die ein Orchestrator wie Airflow ausführt, um eine Pipeline laufen zu lassen —, die sich nur in einem Tabellennamen unterscheiden. Beheben Sie einen Bug, und Sie beheben ihn in allen, wobei Sie einige übersehen. Reagieren Sie darauf mit einer großen Pipeline, die für jeden Fall, dem sie je begegnet ist, ein Konfigurations-Flag trägt, und Sie haben das Problem verschoben statt gelöst: jetzt kann niemand mehr den Executor ändern, ohne einen Fachbereich drei Konten weiter zu zerbrechen. Der Zug, der funktioniert, ist derselbe wie bei den Screens — ändern Sie, was eine Pipeline ist. Deklarieren Sie Quelle, Schritte, Ziel und Policy als Daten; behalten Sie einen kleinen generischen Executor, der die Deklaration interpretiert.
Der Test, ob die Grenze an der richtigen Stelle liegt, ist ebenso konkret. Migriert man einen bestehenden handgeschriebenen Job auf den Deskriptor, muss dabei eine byte-identische Ausgabetabelle und ein identisches Lineage-Ereignis herauskommen. Null Diff, sonst stimmt die Abstraktion nicht.
Ausgehend — die Metadaten, die eine Pipeline emittiert
Der strengere Test auf dieser Seite: jeder Pipeline-Job emittiert, als erstklassige Ausgaben neben seinen Daten, drei Metadaten-Ströme, die nachgelagerte Governance ohne Nachfragen konsumieren kann — Lineage, Assertion-Ergebnisse und Provenance.
Teil 2 behandelt alle drei in der Tiefe: OpenLineage als vendor-neutrale Form, DataZone als AWS-native Senke, das Datenqualitäts-Facet und das Provenance-Facet, aus dem Audit-Berichte erzeugt werden. Hier zählt nur die Richtung. Eine Pipeline, die alle drei als erstklassige Ausgaben emittiert, erzeugt den Nachweis, den die Governance-Schicht braucht, umsonst. Eine Pipeline, die später gescannt, erraten oder rückwärts erschlossen werden muss — von einem Discovery-Tool, das Query-Logs durchkämmt —, verwandelt jenen Nachweis in Dauerarbeit: eine Rekonstruktion, die bei jeder Drift der Plattform wiederholt werden muss und nie ganz fertig wird.
Eine Unterscheidung lohnt sich, bevor Teil 2 sie ausführt, denn das Vermischen der beiden Seiten ist der häufigste Fehler in Qualitätsprogrammen: die Regeln sind eingehende Metadaten, die Ergebnisse sind ausgehend. Eine Expectation Suite ist etwas, das die Pipeline liest. Ein Assertion-Ergebnis ist etwas, das sie produziert. Ein Qualitätsprogramm, das nur das Erste hat, hat eine Richtlinie, deren Anwendung niemand nachweisen kann.
Die Nachrüst-Form
Die Alternative ist das, womit die meisten Plattformen enden, und sie scheitert in beide Richtungen. Ausgehend durchsucht ein Discovery-Tool CloudTrail, errät Lineage aus Query-Mustern und produziert einen Graphen, der teilweise falsch und dauerhaft hinterher ist — ohne Ihnen zu sagen, welche Teile. Eingehend steht die Spaltenliste inline in jedem Job, sodass eine Schemaänderung an der Quelle zu einem Repo-weiten Grep wird — und das ist noch der gute Ausgang, denn jene Jobs sind wenigstens gescheitert. Der schlechte ist der Job, der weiterläuft gegen eine Spalte, die nicht mehr bedeutet, was sie bedeutete, und plausibel aussehende falsche Werte nachgelagert schreibt, so lange, bis es jemandem auffällt. Dann ist die Abhilfe keine Codeänderung mehr, sondern ein Backfill, und ein Backfill funktioniert nur, solange die Rohdaten noch existieren. Sobald die Lifecycle-Regeln der Landing Zone das betroffene Fenster abgelaufen haben, ist die korrumpierte Ausgabe schlicht das, was Sie haben, und der ehrliche nächste Schritt ist Offenlegung statt Reparatur. Es liegt nicht daran, dass die Tools schlecht wären. Es liegt daran, dass die Wahrheit nie irgendwo deklariert wurde, wo eine Maschine sie lesen kann, sodass alles Nachgelagerte eine Form rekonstruiert, die nur der ursprüngliche Autor des Jobs je kannte.
Metadata-first behebt beides, indem es die Quelle der Wahrheit aus dem Code heraus verlagert — in einen Deskriptor, den die Pipeline liest, und einen Ereignisstrom, den die Pipeline emittiert.
§ 2 — Anatomie eines Job-Deskriptors
Hier ist eine echte Transform-Pipeline, deklariert. Das ist das gesamte Artefakt, das ein Fachbereichs-Engineer schreibt, um ein Datenprodukt hinzuzufügen — keine DAG-Datei, keine Job-Klasse, kein 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-interestSechs Blöcke, von denen jeder genau eine Frage beantwortet: woher die Daten kommen, was mit ihnen zu tun ist, wohin sie gehen, was über sie wahr sein muss, gegen welche Regeln sie geprüft werden und auf welcher Rechtsgrundlage sie überhaupt verarbeitet werden.
Zwei Details wiegen schwerer, als sie aussehen:
Das einzig Ausführbare im Deskriptor ist SQL, über einen Pfad referenziert. sql/score.sql enthält die Geschäftslogik des Fachbereichs, und das soll es auch — das ist der Teil, der sich zwischen Pipelines tatsächlich unterscheidet und der tatsächlich dem Fachbereich gehört. Alles andere ist Deklaration. Die Linie zwischen beidem ist die Linie zwischen dem, was der Fachbereich besitzt, und dem, was die Plattform besitzt.
Der processing-Block ist Compliance-Metadatum, das neben dem Code lebt, den es beschreibt. purpose und legal_basis sind DSGVO-Art.-30-Felder. Sie werden hier deklariert, reisen auf dem Lineage-Ereignis nach außen und landen im erzeugten Verarbeitungsverzeichnis — weshalb der Behaviour-first-Governance-Begleitbeitrag ein RoPA projizieren kann, ohne dass ein Compliance-Team eine Tabelle pflegt. Der Pipeline-Autor deklariert die Rechtsgrundlage einmal, in der Datei, die er ohnehin gerade bearbeitet, und der Eintrag pflegt sich selbst.
Dasselbe Schema, eine andere Pipeline
Ein Deskriptor-Format, das nur zu der Pipeline passt, um die herum es entworfen wurde, ist bloß eine Konfigurationsdatei. Hier ist eine Ingestion-Pipeline in einem anderen Fachbereich — andere Quelltechnologie, anderes Ausgabeformat, andere Fehler-Policy —, ausgedrückt im identischen 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 basisNichts am Executor ändert sich, um dies auszuführen. Vier Felder unterscheiden sich im Wert — source.kind, target.kind, expectations.on_failure und processing.legal_basis — und jedes davon ist ein Schlüssel in eine Registry oder ein Enum, das die Plattform bereits implementiert. Genau das kauft die Abstraktion: die zweite Pipeline kostet eine YAML-Datei, und die vierzigste ebenso.
Wie man das Deskriptor-Schema definiert
Das Schema selbst ist die Entwurfsentscheidung, von der alles Übrige erbt. Sechs Regeln erzeugen eines, das den Kontakt mit echten Fachbereichen übersteht:
- Absicht deklarieren, nicht Mechanik.
kind: icebergsagt, was die Ausgabe ist.spark.sql.sources.partitionOverwriteMode: dynamicsagt, wie eine bestimmte Engine das erreicht — und bindet Sie für immer an diese Engine. Jedes Mechanik-Detail, das Sie in den Deskriptor lassen, ist eine Migration, die Sie später bezahlen. - Code über Pfade referenzieren, niemals inline.
query_file: sql/score.sqlhält SQL in einer.sql-Datei, wo eine IDE es linten, ein Reviewer es im Diff lesen und ein Formatter es anfassen kann. YAML-Block-Skalare mit SQL darin sind der Weg, auf dem Deskriptoren zu unlesbaren Programmen verrotten. - Was pro Pipeline variiert, gehört in den Deskriptor; was konstant ist, in den Executor. Das ist der ganze Test, und es lohnt sich, ihn wörtlich anzuwenden. Hat ein Feld in jedem Deskriptor denselben Wert, ist es Executor-Konfiguration im Deskriptor-Kostüm — löschen Sie es.
- Kein Feld darf von seinem Autor verlangen, die Implementierung des Executors zu kennen. Ein Fachbereichs-Engineer schreibt
on_failure: quarantine, ohne zu wissen, wo der Quarantäne-Bucket liegt oder welche IAM-Rolle dorthin schreibt. In dem Moment, in dem ein Feld Implementierung durchsickern lässt, ist der Fachbereich an die Interna der Plattform gekoppelt und die Eigentumsteilung aus § 7 bricht zusammen. - Das Schema ab dem ersten Deskriptor versionieren.
apiVersion: cordata.tech/v1kostet am ersten Tag eine Zeile und ist das, was ein künftigesv2neben v1 landen lässt statt an seiner Stelle — § 3'sload()wählt damit das Modell, gegen das geparst wird, sodass die Migration der Fachbereiche nie zum Stichtag wird. - In der CI gegen dasselbe Modell validieren, das der Executor zur Laufzeit benutzt. Ein Pydantic-Modell, importiert von beiden. Zwei Implementierungen derselben Regeln sind zwei Implementierungen, die auseinanderdriften, und die in der CI driftet immer in Richtung nachsichtiger.
Die Regel mit der meisten Wirkung ist (3). Sie ist auch die, die Teams als Erstes brechen, meist indem sie ein Feld für einen echten Einzelfall hinzufügen. Wenn das passiert, ist der ehrliche Zug die Ausweichluke aus § 3 — nicht ein neues Feld, das jeder Deskriptor trägt und eines benutzt.
Zur Schreibweise, da sie geliehen und der Erwähnung wert ist: apiVersion-Gruppe plus kind ist Kubernetes' Konvention, hier gewählt, weil jeder Plattform-Engineer sie parst, ohne dass man es ihm sagen muss, und weil das Gruppenpräfix verhindert, dass Deskriptoren unterschiedlicher Herkunft kollidieren. Die Abweichung ist bewusst. Kubernetes verschachtelt den Körper unter spec:, weil ein Controller status: zurückschreibt; an einen Deskriptor schreibt nichts zurück, sodass ein spec:-Wrapper eine Einrückungsebene wäre, die einen Status andeutet, der nie eintrifft. Und kind ist heute das Feld mit der geringsten Arbeit — beide Kinds validieren derzeit gegen dasselbe Modell. Es ist da, um ein Verzeichnis von Deskriptoren auf einen Blick überschaubar zu halten und um den Diskriminator-Platz zu reservieren für den Moment, in dem die beiden Formen wirklich auseinandergehen — was sie tun werden, sobald eine Ingestion-Pipeline einen CDC-Modus braucht, für den eine Transformation keine Verwendung hat.
§ 3 — Der generische Executor
Der obige Deskriptor ist träge ohne etwas, das ihn interpretiert. Dieses Etwas enthält keine Tabellennamen, keine Spaltenlisten, kein SQL und keinerlei Fachbereichslogik — alles, was sich zwischen Pipelines unterscheidet, liegt bereits im Deskriptor. Übrig bleibt Dispatch, und Dispatch wächst nicht, wenn ein Fachbereich ein Produkt hinzufügt.
Beginnen wir mit der Form, die ein Deskriptor erfüllen muss. Sie gehört der Plattform, ist versioniert und wird in der CI validiert:
# 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 verdient zuerst ein Wort, weil die Grenze, die es zieht, leicht falsch zu ziehen ist. Eine Quelle und ein Ziel sind beide eine Katalog-Adresse — eine Datenbank und eine Tabelle —, insofern ist das wirklich ein Konzept, und es herauszuziehen gibt beiden ein gemeinsames fqn, das die Reader, das Lauf-Log und Teil 2's Lineage-Payload alle brauchen. Bewusst nicht herausgezogen wird kind: es steht auf beiden, identisch geschrieben und identisch typisiert, aber auf einer Quelle ist es ein Schlüssel in READERS und auf einem Ziel in WRITERS. Gleicher Name, verschiedene Registries — es hochzuziehen wäre geteilte Schreibweise statt geteilter Bedeutung, jene ordentlich aussehende Vererbung, die ein Schema später schwerer lesbar macht.
Drei weitere Entscheidungen lohnen das Hervorheben, denn sie sind es, die das Schema tragen lassen statt bloß parsen.
extra="forbid" macht aus einem vertippten Schlüssel einen gescheiterten Pull Request. Die Voreinstellung — unbekannte Felder still zu ignorieren — ist der Weg, auf dem ein Deskriptor ein Jahr lang on_fail: block trägt, während die Pipeline leise mit der Default-Policy läuft. legal_basis ist ein Literal der sechs Art.-6-Rechtsgrundlagen statt eines freien Strings, sodass eine erfundene Grundlage das erzeugte Verarbeitungsverzeichnis nicht erreichen kann. Und der exactly_one_body-Validator ist die Art Regel, die ins Modell gehört statt in den Executor: query_file und module gemeinsam gesetzt zur PR-Zeit zu fangen, kostet einen Reviewer nichts, während dasselbe zur Laufzeit einen gescheiterten Nachtlauf kostet.
Dann der Executor selbst. Jeder nummerierte Kommentar markiert eine eingehende Metadaten-Art aus § 1, in der Reihenfolge — und der Zweig in Schritt 4 ist der Ort, an dem alle drei on_failure-Policies ihre Bedeutung bekommen:
# 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 2Eine Eigenschaft dieser Form lohnt es, als Regel festzuhalten, denn sie ist der Unterschied zwischen einer Plattform, die für sich selbst Auskunft geben kann, und einer, die es nicht kann: jeder Lauf emittiert genau ein terminales Ereignis. Erfolg, Quarantäne oder Fehlschlag — die Governance-Schicht erfährt von allen dreien.
Das allein aus Disziplin zu bekommen, funktioniert nicht. Drei handgeschriebene emit()-Aufrufe an drei Ausgängen sind drei Gelegenheiten zu vergessen, und der vierte Ausgang, den jemand nächstes Jahr hinzufügt, wird vergessen. Also kommt es stattdessen aus der Struktur: execute wirft, run fängt an einer einzigen Grenze, und jeder Executor-Fehler erbt von PipelineError, sodass ein except alle abdeckt. Eine Pipeline, die an Schema-Drift stirbt, produziert damit weiterhin ein FAIL-Ereignis mit dem Grund — statt eine stille Lücke im Lineage-Graphen zu hinterlassen, wo ein Lauf sein sollte, und das ist genau die Form, die ein Aufsichtsführender nicht von „niemand hat es eingeplant“ unterscheiden kann.
Die Hierarchie, die das einzelne except möglich macht, ist klein genug, um sie vollständig zu zeigen, und lebt in einem eigenen Modul, damit catalog.py und policy.py werfen können, ohne run.py zu importieren, das sie bereits importiert:
# 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 resultNur PublishBlocked hat ein result, das es dem terminalen Ereignis übergeben kann, denn es ist der einzige Fehler, der weit genug kam, um die Suite auszuführen. Der Rest erbt den None-Default, weshalb emit() Validation | None akzeptiert — ein Lauf, der an Schema-Drift starb, produziert immer noch ein Ereignis, es hat nur einen Fehler zu melden statt eines Assertion-Ergebnisses.
Die ehrliche Kante: UnknownApiVersion wird innerhalb von load() geworfen, bevor es einen validierten Deskriptor gibt, dem sich ein Ereignis zuordnen ließe. Für eine Datei, die der Executor nicht parsen kann, lässt sich nichts Sinnvolles emittieren — jener Fehler gehört der CI, die den Deskriptor zur Pull-Request-Zeit genau deshalb validiert hat, damit er nie einen Scheduler erreicht.
Das ist die gesamte Implementierung, abzüglich der Reader- und Writer-Körper. Jeder Deskriptor über alle Fachbereiche hinweg läuft durch diese vierzig Zeilen.
Die Registries sind der Ort, an dem der Executor wächst, und der Vertrag, den sie auferlegen, ist bewusst eng — ein Reader nimmt eine Source und ein aufgelöstes Schema und gibt einen Frame zurück, der diesem Schema entspricht, oder wirft:
# 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)Beide Reader enden bei schema.enforce, und das ist der Vertrag: jeder Reader gibt einen Frame zurück, der dem gepinnten Schema entspricht, sodass nichts Nachgelagertes — kein Schritt, keine Suite, kein Writer — sich gegen eine quellenförmige Überraschung wehren muss. Die Unordnung von Change-Data-Capture aus AWS DMS — dem Database Migration Service, der eine Quelldatenbank stapelweise nach S3 repliziert, ein Stapel Inserts, Updates und Deletes nach dem anderen — bleibt in zwölf Zeilen eingeschlossen, wo collapse_cdc eine schlichte testbare Funktion ist statt eines Blocks, der in jeden Job kopiert wird, der aus DMS liest.
Das gehört vom dedupe_cdc-Schritt getrennt, den der claims-ingest-Deskriptor in § 2 deklariert — die beiden klingen nach derselben Aufgabe und sind es nicht. Der Reader verdichtet das Änderungsprotokoll auf eine Zeile pro Schlüssel: DMS-Mechanik, wahr für jede Quelle, die so ankommt, und damit Sache der Plattform. Der Schritt entfernt die Duplikate im Sinne des Fachbereichs, etwa denselben Schadenfall, zweimal unter zwei Ids gemeldet — was nur das Policy-Team definieren kann. Die Eigentumslinie aus § 7 verläuft genau dazwischen.
Eine Quelltechnologie hinzuzufügen ist ein Eintrag in READERS plus eine Funktion, die den Vertrag einhält — keine neue Pipeline, und kein neues Flag auf einer bestehenden.
Diese Registries sehen nach Konfiguration aus, warum sind sie also Literale? Weil ihre Werte Funktionen sind und Konfiguration nur einen String halten kann, der eine benennt. Sie herauszuziehen bedeutet Importpfade in YAML — ein Plugin-System im Konfigurations-Kostüm, das einen Importfehler beim Modul-Laden in einen fehlgeschlagenen Lookup mitten im Lauf innerhalb eines Fachbereichs-Kontos verwandelt. Es bringt auch nichts, denn die Menge der Reader ändert sich nur, wenn jemand einen Reader schreibt — was ohnehin Code ausliefert. Wieder Entwurfsregel 3, diesmal auf den Executor gerichtet: welche Reader existieren, ist eine Eigenschaft der Executor-Version, nicht irgendeiner Pipeline.
Die Ausnahme ist ein Fachbereich mit einer proprietären Quelle, die sonst niemand liest, wo ein Plattform-Pull-Request die falsche Kopplung wäre; Python Entry Points sind dort der Mechanismus, zu genau jenem Kosten der verzögerten Fehlermeldung. Eine ehrliche Lücke: die python-Schrittart unten deckt eigene Transformationen ab, nicht eigene Quellen, denn sie läuft auf einem bereits gelesenen Frame. Eine exotische Quelle ist heute weiterhin eine Plattform-Änderung.
Ein Lauf gegen den Transform-Deskriptor aus § 2 liest sich so:
$ 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
Jede Zeile entspricht einer der fünf eingehenden Metadaten-Arten, die aufgelöst wird, oder dem ausgehenden Ereignis, das diese Auflösung ermöglicht. Nichts in dieser Ausgabe ist spezifisch für Fraud-Scoring. Richten Sie denselben Befehl auf domains/policy/pipelines/claims_ingest.yml, und die Form ist identisch — anderer Reader, anderer Writer, quarantine statt block_publish und eine contract-Rechtsgrundlage auf dem emittierten Ereignis.
Der Fehlerpfad ist der lehrreichere. Wenn das Upstream-Team eine Spalte zu fraud_raw.transactions hinzufügt:
[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
Nichts wurde gelesen, nichts geschrieben, nichts veröffentlicht, und der Fehler benennt die exakte Spalte und die exakte Abhilfe. § 4 erklärt, warum das das richtige Verhalten ist und keine Unannehmlichkeit.
Wo der generische Executor aufhört zu funktionieren
Jede Abstraktion hat eine Grenze, und so zu tun, als hätte diese keine, ist der Weg, auf dem man auf einem anderen Umweg bei derselben überkonfigurierten Pipeline ankommt.
Drei Formen widersetzen sich der Deklaration wirklich: Transformationen, deren Logik sich nicht als geordnete SQL-Schritte ausdrücken lässt (iterative Algorithmen, Graph-Traversierungen, alles mit echtem Kontrollfluss); Quellen, die sich nicht wie Tabellen verhalten, etwa eine Vendor-API, die eigene Pagination und eigenes Retry-Handling braucht; und einmalige Backfills, die zweimal laufen und nie wieder.
Die Ausweichluke ist die python-Schrittart, die auf ein fachbereichseigenes Modul zeigt:
steps:
- id: reconcile
kind: python
module: domains.fraud.transforms.reconcile_late_settlements
params:
lookback_days: "30"Der Sinn, die Ausweichluke durch den Deskriptor zu führen statt an ihm vorbei: das Modul ist beliebiges Python, aber die Pipeline deklariert weiterhin ihre Quelle, ihr Ziel, ihren Vertrag, ihre Expectation Suite und ihre Rechtsgrundlage — sodass Lineage, Assertions und Tag-Propagation alle weiterhin stattfinden. Die eigene Logik ist auf die eine Stelle beschränkt, an der sie wirklich gebraucht wird, und der Metadaten-Vertrag bleibt intakt. Eine Ausweichluke, die den Executor ganz umgeht, würde all das verwirken — was genau das ist, was eine handgeschriebene DAG tut.
Wann man das gar nicht erst bauen sollte
Unterhalb von etwa zwanzig Pipelines kostet der Executor mehr, als er spart. Das Deskriptor-Schema, die CI-Validierung, die Reader- und Writer-Registries, die Dokumentation, die ein Fachbereichs-Engineer braucht, um seinen ersten Deskriptor zu schreiben — das ist eine Plattform-Investition, und sie zahlt sich über Wiederholung aus. Eine Pipeline von Hand zu schreiben ist nicht das Problem. Die vierzigste von Hand zu schreiben schon.
§ 4 — Schemata und Verträge, aus dem Katalog aufgelöst
schema_version: 7 ist die folgenreichste Zeile im Deskriptor, und ihr ganzes Gewicht liegt in der Zahl.
Der Executor löst sie zur Laufzeit gegen den Glue Data Catalog auf:
# 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 nimmt die ganze Source statt eines Referenz-Strings, weshalb hier überhaupt nicht geparst wird. Ein früherer Entwurf dieses Schemas adressierte die Tabelle mit einer URI — catalog://fraud_raw/transactions@v7 — und das erwies sich als kleine selbstzugefügte Wunde: der Deskriptor deklariert database und table bereits, sodass die URI beides wiederholte und zwei Stellen hinterließ, die man im Gleichschritt halten musste. Entwurfsregel 3 hat es gefangen. Die Version war das Einzige, was die Referenz wirklich trug, also bleibt die Version das Einzige, was übrig ist.
Das interessante Verhalten ist der Fehlschlag. Wenn ein Upstream-Team eine Spalte hinzufügt, rückt die Quelltabelle auf v8, der Deskriptor pinnt weiterhin v7, und der Lauf scheitert laut mit einem Diff auf Spaltenebene — bevor irgendetwas geschrieben wird. Der Fachbereichs-Engineer liest dann den Diff, entscheidet, ob die neue Spalte relevant ist, und hebt den Pin in einem Pull Request an. Dieser PR ist der Nachweis der Entscheidung.
Vergleichen Sie das mit den zwei üblichen Alternativen. SELECT * absorbiert die neue Spalte still und reicht sie nachgelagert weiter, wo das Schema-on-Read eines Konsumenten eine Woche später ohne erkennbare Ursache bricht. Eine fest verdrahtete Spaltenliste ignoriert sie still, was in Ordnung ist, bis das Upstream-Team etwas umbenennt und die Pipeline weiterläuft gegen eine Spalte, die nicht mehr bedeutet, was sie bedeutete. Keine der beiden erzeugt einen Moment, in dem ein Mensch etwas entscheidet.
Die Zielseite funktioniert genauso, nur umgekehrt. Der contract-Block ist das, was die Pipeline ihren Subscribern verspricht — Ausgabeschema, Frische-SLA und die LF-Tags, die das Produkt trägt. Er treibt, was der Writer emittiert, und was beim Publish im Katalog registriert wird, aus einer Deklaration. Subscriber, die den Katalog lesen, lesen denselben Vertrag, gegen den die Pipeline ausgeführt wurde, und keine Wiki-Seite, die ihn beschreibt.
Data Catalog, nicht Schema Registry — und warum das den Entwurf begrenzt
Der Präzision halber, denn AWS liefert zwei Dienste mit „Glue“ und „Schema“ im Namen, die verschiedene Probleme lösen.
Der Glue Data Catalog ist ein Metastore für ruhende Daten: Datenbanken, Tabellen, Spalten, Partitionen und eine lineare Versionshistorie jeder Tabellendefinition. Ihn lesen Athena, EMR, Redshift Spectrum, LakeFormation und die Glue-Jobs in diesen Beiträgen alle, und ihn adressieren database, table und schema_version eines Deskriptors oben. Die Glue Schema Registry ist ein separater Dienst für Daten in Bewegung — Avro-, JSON-Schema- und Protobuf-Nachrichtenschemata für MSK, Kinesis und Flink — und ihr Unterscheidungsmerkmal ist, dass sie Kompatibilitätsregeln bei der Registrierung durchsetzt.
Dieser Unterschied ist nicht kosmetisch, und er legt die ehrliche Begrenzung des Obigen offen. Eine Registry weist ein inkompatibles Schema zurück, wenn der Produzent es zu registrieren versucht, bevor eine einzige Nachricht ausgeliefert wird. Der Data Catalog hat keinen entsprechenden Moment zum Einhaken: eine Tabellendefinition ändert sich, wann immer das Upstream-Team DDL ausführt oder ein Crawler durchläuft, und die erste Partei, die es bemerken könnte, ist die, die als Nächstes liest. Laufzeit-SchemaDrift ist deshalb nicht der ideale Ort, um Drift zu fangen — es ist der früheste verfügbare für ruhende Tabellen.
Streaming-Quellen sollten diese Prüfung nach links in die Registry verschieben, statt sie pro Konsument neu zu entdecken, was einer der konkreten Gründe ist, warum die Streaming-Form eine eigene Behandlung braucht statt eines Absatzes hier. Ein Deskriptor für eine Streaming-Quelle würde einen Kompatibilitätsmodus benennen statt eine Katalogversion zu pinnen — ein anderer Mechanismus für dieselbe Absicht.
§ 5 — Expectation Suites als deklarierte Eingabe
Assertions sind der eine eingehende Strom mit einer echten Werkzeugwahl daran, deshalb lohnt es sich zu sagen, warum diese Architektur zu Great Expectations greift statt zur naheliegenderen Option.
Die naheliegende Option ist, wenn die Transform-Ebene dbt ist, dbt's eigene Tests — not_null, unique, accepted_values, relationships, deklariert im selben YAML wie die Modelle, erweitert um dbt-expectations, wo die eingebauten nicht reichen. Für einen Shop, der Ende zu Ende dbt ist, ist das die richtige Antwort, und ein zweites Assertion-Framework hinzuzufügen ist ein Aufwand ohne Gegenwert.
Die tiefere Grenze ist allerdings nicht, welche Pipelines dbt ausführen — es ist, welche ETL-Stufe dbt sehen kann. dbt ist das T. Es kann etwas weiter reichen — sources: mit Tests ist ein echtes Feature, sodass frisch gelandete Tabellen ebenfalls geprüft werden können —, aber es prüft, was bereits in einem Warehouse angekommen ist, das es abfragen kann, in einem eigenen Aufruf. Das E und das L sind damit nachträglich abgedeckt, wenn überhaupt, und eine abgeschnittene Quelldatei oder ein halb angewendeter CDC-Stapel ist bereits gelandet, bevor irgendetwas dagegen läuft.
Dieselbe Lücke wiederholt sich dann pro Pipeline. Die Schritte eines Deskriptors dispatchen über die STEPS-Registry — heute SQL, ein Python-Modul für die Ausnahmen aus § 3, was ein Fachbereich als Nächstes registriert — und der claims-ingest-Deskriptor aus § 2 berührt dbt an keinem Punkt. Die Regel, die die Plattform tatsächlich braucht, ist, dass jedes Datenprodukt vor der Veröffentlichung geprüft wird, egal welche Engine es produziert hat und auf welcher Stufe, und kein Werkzeug, das eine Engine und eine Stufe sieht, kann sie durchsetzen.
Great Expectations kauft genau diese Reichweite: es validiert einen Frame, kein Warehouse, sodass dieselbe Suite-Form hinter jedem Reader, bei jedem Schritt, in jeder Pipeline funktioniert. Seine Ergebnisse kommen außerdem als strukturierte Objekte statt als Log-Zeilen, was es § 3's validate erlaubt, emit() ein einheitliches Assertion-Ergebnis zu übergeben, egal was gelaufen ist. Die Kosten sind ehrlich — ein zweites Framework, das gelernt werden will, und statistische Expectations, die eine gespeicherte Baseline zum Vergleich brauchen.
Deshalb versteht man es hier besser als eingehenden Metadaten-Strom denn als Test-Werkzeug: ein versioniertes Artefakt, das die Pipeline liest, dessen Assertion-Ergebnisse sie emittiert.
Die Suite liegt im Repository des besitzenden Fachbereichs, neben dem Deskriptor, der sie referenziert, und dem SQL, das jene Schritte ausführen. Jeder Pfad innerhalb eines Deskriptors — sql/…, expectations/… — löst gegen diese Fachbereichs-Wurzel auf:
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}Die ersten drei sind Hygiene, die Sorte Prüfung, die ein Typsystem umsonst mitliefern würde. Die letzten beiden sind anderer Art. Verteilungs-Drift auf dem Fraud-Score — KL-Divergenz gegen eine gespeicherte Baseline — fängt eine Modell-Regression, die jeder Schema-Test der Welt bestehen würde. Zeilenzahl-Grenzen fangen Lösch-Bugs stromaufwärts, bevor sie in das Dashboard eines Konsumenten kaskadieren.
partition_object: baseline_fraud_score benennt diese Baseline, statt sie einzubetten. Der Executor löst den Namen gegen ein baselines/-Verzeichnis im selben Fachbereichs-Repository auf. Das hält hundert Bin-Gewichte aus einer Datei heraus, die Menschen lesen müssen — und es sorgt dafür, dass ein Neu-Baselining als Diff mit einem Autor daran ankommt statt als Zahl, die sich still in einer Suite ändert. „Das Modell ist gedriftet“ und „wir haben entschieden, dass die neue Verteilung in Ordnung ist“ dürfen in der Historie nicht gleich aussehen.
Die Platzierung zählt genauso viel wie der Inhalt. Die Suite läuft an der Pipeline-Grenze — nach den deklarierten Schritten, vor dem Schreiben. Nicht verstreut über die Transformation. Nicht auf einem geplanten Scan Stunden später. Genau an der Naht, an der die Pipeline gleich sagen wird „hier ist die Ausgabe, die ich meinen Subscribern verspreche“, wird das Versprechen geprüft. Deshalb ist on_failure: block_publish eine sinnvolle Einstellung: an jener Naht bedeutet Scheitern, dass nichts veröffentlicht wird und kein Konsument je den schlechten Batch sieht.
Die drei on_failure-Modi existieren, weil nicht jede Regel dieselbe Reaktion verdient. block_publish für Korrektheits-Invarianten — ein Null-Primärschlüssel sollte die Welt anhalten. quarantine für Volumen-Anomalien, wo an einen Nebenort zu schreiben und den owner des Deskriptors zu alarmieren besser ist als beides, Veröffentlichen und Anhalten. warn für Regeln, die noch justiert werden, was die ehrliche Einstellung für eine neue Verteilungsgrenze ist, deren Schwelle noch niemand traut. Eine Suite, in der alles blockiert, wird binnen eines Monats abgeschaltet.
Eigentum folgt derselben Linie wie beim Deskriptor: der Fachbereich besitzt die Suite, weil der Fachbereich die einzige Partei ist, die weiß, wie fraud_score aussehen soll. Die Plattform besitzt den Runner. Eine Regel zu ändern ist ein Pull Request gegen das Repository des Fachbereichs — was bedeutet, dass die Qualitäts-Policy eine Commit-Historie, einen Autor und einen Reviewer hat, statt ein Absatz in einem Governance-Dokument zu sein.
Was die Suite produziert — das Assertion-Ergebnis pro Lauf und was Governance damit tut — ist Teil 2.
§ 6 — Policy-Metadaten und LF-Tag-Auflösung
Der Deskriptor benennt Tag-Werte:
contract:
lf_tags:
sensitivity: high
residency: eu
subject_type: customerEr definiert nicht, was high bedeutet, was es erlaubt oder wer Daten lesen darf, die es tragen. Dieses Vokabular — die LF-Tag-Ontologie aus Fabric + Mesh auf AWS — liegt im Governance-Konto und gehört der Governance-Funktion. Der Executor löst zur Laufzeit dagegen auf:
# 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 declaredZwei Eigenschaften fallen dabei ab:
Umklassifizierung propagiert, ohne Pipelines anzufassen. Wenn Governance entscheidet, dass residency: eu nun einen engeren Grant-Ausdruck impliziert, geschieht die Änderung einmal, in der Ontologie. Jede Pipeline, die ihre Ausgabe mit residency: eu taggt, erbt die neue Bedeutung beim nächsten Lauf. Das ist der Unterschied zwischen Policy-as-Code und Policy-als-Spalte-die-Sie-kopiert-haben.
Unbekannte Werte scheitern, bevor sie Produktion erreichen. Dieselbe Validierung läuft in der CI gegen den Deskriptor, sodass ein Tippfehler wie sensitivity: hgh zur Pull-Request-Zeit mit der erlaubten Menge in der Fehlermeldung scheitert — nicht gegen eine Tabelle, die bereits mit einem Tag geschrieben wurde, gegen das nichts grantet. Dieselbe Regel an beiden Stellen aus derselben Ontologie zu validieren ist das, was die CI-Prüfung vertrauenswürdig macht statt zu einer zweiten Implementierung, die driftet.
§ 7 — Eigentum, Ablage und der Standard, den es nicht gibt
Eigentum teilt sich dreifach, und es bildet genau den Drei-Wege-Vertrag aus dem Governance-Grundpfeiler ab:
- Die Plattform besitzt den Executor und das Deskriptor-Schema — die Form, die jeder Deskriptor erfüllen muss.
- Der Fachbereich besitzt die Deskriptor-Instanzen für seine eigenen Datenprodukte, das SQL, das jene Deskriptoren referenzieren, und seine Expectation Suites.
- Governance besitzt das Policy-Vokabular, das jene Deskriptoren referenzieren — die LF-Tag-Ontologie, die Sensitivitätsstufen, die Rechtsgrundlagen-Werte.
Diese Teilung ist der Grund, warum das Modell unter einem föderierten Betriebsmodell trägt, statt sich still wieder zu zentralisieren. Das Plattform-Team reviewt nie die Geschäftslogik eines Fachbereichs; der Fachbereich bearbeitet nie den Executor; Governance fasst nie eine Pipeline an. Jede Seite ändert nur die Artefakte, die ihr gehören, und das Deskriptor-Schema ist der Vertrag zwischen ihnen.
Die Ablage folgt aus dem Eigentum, und was Teams stolpern lässt, ist, dass ein Deskriptor drei Zuhause hat, die leicht zu verwechseln sind:
- Quelle der Wahrheit — das Git-Repository des besitzenden Fachbereichs. Kein AWS-Konto. Der Deskriptor, das SQL, das seine Schritte referenzieren, und die Expectation Suite liegen zusammen unter den CODEOWNERS jenes Fachbereichs und werden wie jede andere Änderung reviewt. Der signierte Commit ist dann bereits der Änderungs-Autorisierungs-Nachweis, den Behaviour-first Governance in der Praxis § 4 beschreibt — dieser Nachweis kostet nichts extra.
- Laufzeit-Kopie — ein Config-Präfix im Producer-Konto. Der Executor läuft dort, wo die Daten sind, also muss der Deskriptor von dort erreichbar sein; die CI des Fachbereichs synchronisiert ihn beim Merge. Diese Kopie ist abgeleitet und wegwerfbar. Niemand bearbeitet sie an Ort und Stelle, und sie aus dem Repository neu zu bauen ist immer sicher.
- Das Schema, das er erfüllen muss — ein versioniertes Paket, das die Plattform veröffentlicht. Importiert von der CI des Fachbereichs und vom Executor, was die Pull-Request-Prüfung vertrauenswürdig macht statt zu einer zweiten Implementierung, die driftet.
Die Schemata selbst liegen im Glue Data Catalog als Laufzeit-Quelle der Wahrheit, gemäß § 4.
Die verlockende Abkürzung ist ein einziges zentrales Repository, das die Deskriptoren aller Fachbereiche enthält und vom Plattform-Team verwaltet wird. Das reproduziert genau den Engpass, den diese Architektur beseitigen soll: jede Pipeline-Änderung wird zu einem Posten in einer Warteschlange, die dem Plattform-Team gehört. Deskriptoren sind Fachbereichs-Artefakte und gehören in fachbereichseigene Repositories — ob das ein Repository pro Fachbereich bedeutet oder ein Monorepo mit CODEOWNERS pro Verzeichnis, zählt weit weniger als die Frage, wer den Merge freigibt.
Wo der Code läuft, folgt derselben Logik. Der Executor läuft innerhalb des Producer-Kontos, neben den Daten, aus genau dem Grund, aus dem die DMS-Instanz es tut: Quell-Zugangsdaten und Rohzeilen überqueren nie die Kontogrenze. Zentral ist das Executor-Artefakt — die Plattform baut und veröffentlicht ein versioniertes Image, jedes Fachbereichs-Konto konsumiert es. Eine Implementierung, viele Konten. Die Orchestrierung ist das eine Stück, das wirklich in die Mitte gehört, und Teil 2 arbeitet durch, warum — einschließlich der Stelle, an der das aufhört zu gelten.
Von Anfang bis Ende sieht das Definieren und Onboarden einer neuen Pipeline so aus:
Diagramm-Lesehilfe. Ein Fachbereichs-Engineer öffnet einen Pull Request mit drei Dateien und keinem Code — Deskriptor, SQL, Expectation Suite. Die CI validiert ihn gegen das Deskriptor-Schema der Plattform und prüft, ob die gepinnte schema_version im Katalog noch die aktuelle ist, sodass ein Tippfehler oder ein veralteter Versions-Pin von einem Reviewer gefangen wird statt von einem Pager. Beim Merge wird der signierte Commit zum Änderungs-Autorisierungs-Nachweis, und die CI synchronisiert den Deskriptor in das Config-Präfix des Producer-Kontos. Beachten Sie, was an dieser Stelle nicht passiert: es wird kein neuer Code gebaut oder deployt, denn der von der Plattform veröffentlichte Executor ist bereits da — eine Pipeline zu onboarden ist eine Konfigurationsänderung, kein Release. Zur Laufzeit liest der Executor das Quellschema aus dem Katalog, führt die deklarierten Schritte aus, prüft die Suite, löst die Policy-Tags auf und emittiert seine drei Ströme an das Governance-Konto — dort setzt Teil 2 an.
Eine ehrliche Asymmetrie: die eingehende Seite hat nichts mit OpenLineages Konsolidierung dahinter. Teile eines Deskriptors sind durch offene Standards abgedeckt — dbt's source- und model-YAML ist die De-facto-Form für die Transform-Ebene, und wo dbt die Engine ist, sollten Sie sie übernehmen statt eine parallele zu erfinden.
Was keiner von ihnen bisher tut, ist Quelle, Schritte, Ziel, Vertrag, Expectations und Policy in einer einzigen Deklaration zu umspannen — und das ist der ganze Grund, warum apiVersion in § 2 cordata.tech/v1 liest statt etwas zu benennen, das einem Standardisierungsgremium gehört. Ein kleiner hauseigener Deskriptor, veröffentlicht als versioniertes JSON Schema, ist die pragmatische Antwort, bis einer jener Teilstandards wächst, um den Rest abzudecken.
Der Trost ist, dass der Deskriptor Ihr Schema ist, in Ihrem Repository, und ihn später zu migrieren eine mechanische Transformation über ein Verzeichnis von YAML-Dateien ist. Das ist eine deutlich bessere Position als vierzig handgeschriebene DAGs auf irgendetwas zu migrieren.
Teil 2 — der Nachweis, den eine Pipeline emittiert — behandelt OpenLineage als vendor-neutrales Lineage-Rückgrat, das AWS-native Adoptions-Playbook über DataZones PostLineageEvent, Assertion-Ergebnisse als Datenqualitäts-Facet und das Provenance-Facet, aus dem Audit-Berichte erzeugt werden.
Der Code in diesem Beitrag läuft. cordata-tech/pipeline-runtime ist der Executor von oben, mit DuckDB und lokalem Parquet anstelle von Glue und Spark. Ein frisch geklontes Repository führt beide Deskriptoren aus § 2 vollständig aus, ohne Cloud-Konto, und reproduziert den SchemaDrift-Fehlschlag aus § 4 aus einer Fixture. Beim Bauen kamen außerdem fünf Stellen ans Licht, an denen diese beiden Beiträge falsch lagen oder nicht wie abgedruckt liefen — sie sind im Repository dokumentiert, die wesentlichen inzwischen hier korrigiert.
Wenn Sie eine deskriptorgetriebene Pipeline-Plattform gebaut haben — oder eine gebaut und festgestellt haben, dass die Abstraktionsgrenze an der falschen Stelle lag — würde ich wirklich gerne hören, wo sie sich gebogen hat. Die Tür steht offen unter cordata.tech/contact. Weiterlesen: der Data-Engineering-Grundpfeiler für das zusammengesetzte Argument, die Fabric + Mesh auf AWS-Referenz für die Account-pro-Fachbereich-Topologie, auf der diese Pipeline sitzt, und Behaviour-first Governance in der Praxis dafür, wie die Governance-Schicht liest, was diese Pipelines produzieren.