Lineage nachvollziehbar machen

Track KI · M6 Datenseite, Baustein 03, Teil 1 · ca. 55 Min.

Worum es geht

Irgendwann steht in einem Bericht eine Zahl, die nicht stimmt. Die eigentliche Frage lautet dann nicht “was ist falsch?”, sondern: woher kam der Wert, und durch welche Schritte ist er gelaufen? Ohne Lineage (Herkunftsnachweis, Datenabstammung) suchst du im Blindflug durch den ganzen Pipeline-Code. Mit Lineage liest du es ab.

Dieser Teil baut drei Dinge, alle im Browser und alle klein:

  1. Lineage als Graph (Datensätze und Transformationen), mit Abfragen nach oben (upstream: woher?) und nach unten (downstream: wer ist betroffen?).
  2. Herkunft pro Zeile mitführen: Quelle, Version, Zeitpunkt und ein Hash als Fingerabdruck.
  3. Reproduzierbarkeit: gleicher Input, gleiche Version, gleicher Hash.

In Teil 2 folgen der Audit-Trail für KI-Antworten, die Projektaufgabe (Abschluss-Check von M6) und ein Rückblick auf den ganzen Lernpfad. Den Stoff von Lektion 18 (Polars, Data Contract, Prüfregeln) wiederholen wir nicht. Du brauchst von dort nur die Idee: Regeln werden aus dem Contract abgeleitet und prüfen Zeilen.

Was im Browser läuft: alles, denn Lineage ist bei uns reines Python (dicts, Listen, hashlib). Echte Lineage-Werkzeuge gibt es hier nicht.

Zeitplan ehrlich: etwa 25 Minuten Lesen, 30 Minuten für die drei Übungen.

Von JS/TS her gedacht

Lineage kennst du aus deinem Alltag, nur heißt es dort anders:

Idee Web / TypeScript Hier
Woher kommt das? (upstream) npm ls, npm why paket, Source Map (Zeile im Bundle zeigt auf Quelldatei) Vorfahren eines Datensatzes im Graphen
Wer ist betroffen? (downstream) “Find usages”, npm ls in umgekehrter Richtung, Dependabot-Hinweis Nachfahren, Impact-Analyse
Gleicher Stand package-lock.json mit Integritäts-Hash, contenthash im Dateinamen Version plus Hash je Datensatz
Reproduzierbarer Build gleicher Commit, gleiches Lockfile, gleiches Bundle gleicher Input, gleiche Version, gleicher Hash
Ablaufspur OpenTelemetry-Spans Span aus Lektion 15, erweitert um Daten-Hashes

Der Unterschied zum Code: Bei Code liegt die Wahrheit im Repository. Bei Daten entsteht sie zur Laufzeit. Eine Zeile in einer Tabelle weiß nicht von selbst, aus welcher Datei sie kam. Wenn du es nicht beim Entstehen mitschreibst, ist die Information weg.

Konzept

Schritt 1: Das Protokoll aus der Quelle

Die Quelle (Baustein 03) zeigt das Minimum: Jeder Transformationsschritt hängt einen Eintrag an ein Protokoll. Dort ist es ein Pydantic-Modell Herkunft mit quelle, schritt und zeitpunkt. Wir nehmen dicts mit denselben Feldern, damit der Code im Browser klein bleibt. Den Zeitpunkt reichen wir als Argument herein, statt datetime.now() im Schritt aufzurufen. Dann ist der Code testbar, und du siehst gleich, warum das für Reproduzierbarkeit wichtig ist.

Das reicht im Kleinen. Zwei Grenzen sind sofort sichtbar: Das Protokoll gilt für den ganzen Lauf, nicht für eine einzelne Zeile. Und es ist eine Liste, kein Netz: Sobald ein Datensatz aus zwei anderen entsteht (Join), brauchst du einen Graphen. Die Quelle sagt dazu: Bei größeren Pipelines übernehmen dedizierte Werkzeuge (z. B. OpenLineage) die Aufgabe automatisiert, mit Visualisierung des Datenflusses. Das Prinzip bleibt gleich, und das Prinzip lernst du hier.

Die Quelle gibt den Merksatz mit: Lineage ist die Datenversion von strukturiertem Logging aus M0 (Lektion 04): nicht erst nachrüsten, wenn etwas schiefgeht, sondern von Anfang an mitführen.

Schritt 2: Lineage als Graph

Ein Graph besteht aus Knoten (Datensätze) und Kanten (“entstand aus”). In Python reicht ein dict: Schlüssel ist der Name eines Datensatzes, der Wert nennt seinen Typ und seine Eingaben (von). Jede Transformation ist dabei die Kante zwischen Eingaben und Ergebnis. Wer die Eingaben kennt, kennt den Graphen. (Das ist dieselbe Darstellung wie die Abhängigkeitsliste in einem Build-System: “A hängt von B und C ab”.)

Als Bild (Mermaid) sieht dasselbe so aus. Pfeile zeigen in Datenflussrichtung:

flowchart LR
  Z[zaehler_export.csv] --> R[messwerte_roh]
  R --> G[messwerte_geprueft]
  G --> T[tagesmittel]
  T --> A[abrechnung]
  F[tarife.csv] --> A
  T --> B1[bericht_netzlast]
  A --> B2[bericht_rechnungen]

Upstream-Abfrage: woher kommt ein Datensatz? Das ist eine Graphsuche über von. Wichtig ist die Menge gesehen: Sie verhindert, dass ein Knoten zweimal bearbeitet wird (Raute im Graphen) und dass die Suche bei einem Zyklus nie endet. Ein Zyklus sollte in Lineage nicht vorkommen (Daten entstehen aus früheren Daten), aber fehlerhafte Metadaten gibt es, und eine Abfrage, die dann hängt, ist schlimmer als ein falsches Ergebnis.

Lies die Ausgabe: Der Rechnungsbericht hängt über die Abrechnung an sechs Datensätzen, darunter beide Quellen. Eine Quelle hat keine Vorfahren, die Liste ist leer.

Ein Zyklus zur Probe, zwei Datensätze, die sich gegenseitig als Eingabe nennen:

Ohne gesehen würde diese Zeile nie fertig. Genau deshalb haben die Checks unten ein Schrittlimit.

Downstream-Abfrage: wer ist betroffen? Der Graph kennt nur die Richtung “entstand aus”. Für die Gegenrichtung drehst du ihn einmal um: Aus “B hat Eingabe A” wird “A hat Nachfolger B”. Danach ist die Suche dieselbe, nur mit dem umgedrehten dict.

Die Tarifdatei fließt nur in die Abrechnung und von dort in den Rechnungsbericht. Der Netzlast-Bericht ist davon nicht betroffen. Das ist eine Impact-Analyse (Auswirkungsanalyse): “Wenn sich Quelle X ändert, welche Berichte muss ich neu prüfen?” Mit einem Filter auf den Typ wird daraus die Frage, die ein Fachbereich wirklich stellt:

Das schreibst du in Übung 1 selbst, mit anderen Daten.

Schritt 3: Herkunft pro Zeile und der Fingerabdruck

Der Graph beantwortet “welcher Datensatz”. Der Alltag fragt aber oft nach einer Zeile: Woher kam genau dieser Zählerwert? Dafür bekommt jede Zeile ein Metadatenfeld mit Quelle, Version (welche Lieferung, welcher Code-Stand), Zeitpunkt und einem Hash.

Ein Hash ist ein Fingerabdruck fester Länge: Gleicher Inhalt gibt immer denselben Hash, schon ein geänderter Wert gibt einen ganz anderen. Das ist die Datenversion des Lockfile-Hashes aus npm. Zwei Dinge musst du dabei beachten:

  1. Der Hash braucht Bytes, also gibst du Daten zuerst in eine feste Textform. Bei einem dict ist das kanonisches JSON: Schlüssel sortiert, feste Trennzeichen. Sonst hätte {"a": 1, "b": 2} einen anderen Hash als {"b": 2, "a": 1}, obwohl beides dasselbe ist.
  2. Python hat eine eingebaute Funktion hash(). Die ist nicht dafür gedacht: Ihr Ergebnis für Strings ändert sich von Programmstart zu Programmstart. Für Fingerabdrücke nimmst du hashlib.

Lies die Ausgabe: z1 und z2 haben denselben Hash, z3 einen anderen. SHA-256 liefert 64 Hex-Zeichen, im Text kürzen wir auf die ersten 8 bis 12 zur Lesbarkeit. Pyodide kann hashlib.sha256. Es fehlen dort nur die Passwort-Verfahren pbkdf2_hmac und scrypt, die du hier nicht brauchst.

Jetzt das Metadatenfeld. Die Funktion baut nur das Herkunftsfeld. Der Hash gilt für den Inhalt der Zeile ohne dieses Feld. Sonst würde das Feld sich selbst enthalten, und jede Änderung am Zeitpunkt würde den Hash der Daten ändern:

Falle bei der Anwendung: Wer zeile["_herkunft"] = ... schreibt, verändert die Eingabe. Wenn dieselbe Zeile in einer anderen Pipeline weiterverwendet wird, steht dort plötzlich ein Feld, das jemand anderes nicht erwartet. Baue eine neue Zeile ({**zeile, "_herkunft": ...}). Das übst du in Übung 2.

Schritt 4: Reproduzierbarkeit

Die Frage: Wenn ich die Pipeline morgen noch einmal laufen lasse, kommt dasselbe heraus? Reproduzierbar heißt: gleicher Input, gleiche Code-Version, gleiches Ergebnis. Mit Hashes lässt sich das prüfen, ohne Daten zu vergleichen: Ein Lauf wird zu drei Werten (Hash der Eingabe, Version, Hash der Ausgabe).

Zeile 1 und 2: gleicher Input, gleiche Version, gleiche Ausgabe. So soll es sein. Zeile 3 zeigt eine Eigenheit: Die Zeilen in anderer Reihenfolge ergeben einen anderen Eingabe-Hash (eine Liste ist geordnet), obwohl das Ergebnis dasselbe ist. “Gleicher Input” muss also kanonisch gemeint sein: sortiere Zeilen vor dem Hashen, wenn ihre Reihenfolge keine Bedeutung hat.

Was Reproduzierbarkeit kaputtmacht:

Der Zeitpunkt im Ergebnis macht jeden Lauf anders. Die Regel: Der Zeitpunkt gehört in die Herkunft (Metadaten), nicht in die Daten, über die du den Hash bildest. Weitere typische Störer: random ohne festen Startwert (Seed), ein Set oder dict, dessen Reihenfolge ins Ergebnis läuft, und eine neue Bibliotheksversion.

Und wenn sich die Version ändert, darf die Ausgabe sich ändern. Das ist kein Fehler, sondern der Grund, warum die Version zum Schlüssel gehört:

Gleicher eingabe-Hash wie oben, neue Version, neue Ausgabe: erwartbar. Auffällig wäre nur dieser Fall: gleiche Eingabe, gleiche Version, aber verschiedene Ausgabe. Das prüfst du in Übung 3.

Falle

  1. Lineage erst nachrüsten. Was beim Entstehen nicht mitgeschrieben wurde, ist später nicht rekonstruierbar. Die Quelle sagt es als Merksatz: von Anfang an mitführen.
  2. Stillschweigen mit “keine Abhängigkeit” verwechseln. Fehlt in der Lineage eine Kante, heißt das nicht, dass es keine gibt. Es heißt vielleicht nur, dass sie niemand eingetragen hat (dasselbe Muster wie “Contract schweigt” in Lektion 18). Die Impact-Analyse ist nur so gut wie der Graph.
  3. Eingabe verändern. zeile["_herkunft"] = ... schreibt in die Zeile, die jemand anderes noch benutzt. Neue Zeile bauen.
  4. Hash über das Herkunftsfeld. Der Hash gilt für den Inhalt ohne _herkunft, sonst ändert jeder neue Zeitpunkt den Fingerabdruck der Daten.
  5. hash() statt hashlib. Der eingebaute hash() ist für Wörterbücher da und ändert sich von Lauf zu Lauf.
  6. Hash über nicht kanonische Daten. Andere Schlüsselreihenfolge, andere Zeilenreihenfolge oder ein Gleitkomma-Rest ergeben einen anderen Hash für “dasselbe”. Lege fest, wie Daten vor dem Hashen aussehen.
  7. Zeitstempel im Ergebnis. Er macht jeden Lauf einzigartig und Reproduzierbarkeit unmöglich. Der Zeitpunkt gehört in die Metadaten.
  8. Version vergessen. Ohne Version vergleichst du Läufe, die legitim verschieden sind, und meldest Fehlalarme. Mit ihr siehst du den echten Verdacht: gleiche Eingabe, gleiche Version, verschiedene Ausgabe.
  9. Zyklen in der Abfrage. Eine Graphsuche ohne Menge besuchter Knoten hängt bei fehlerhaften Metadaten für immer.

Übungen

Übung 1: Impact-Analyse (leicht bis mittel)

Ein Online-Shop hat diese Lineage (dasselbe Format wie im Text, andere Daten):

GRAPH = {
    "crm_export":        {"typ": "quelle",  "von": []},
    "shop_log":          {"typ": "quelle",  "von": []},
    "kunden":            {"typ": "tabelle", "von": ["crm_export"]},
    "bestellungen":      {"typ": "tabelle", "von": ["shop_log", "kunden"]},
    "bericht_umsatz":    {"typ": "bericht", "von": ["bestellungen"]},
    "bericht_neukunden": {"typ": "bericht", "von": ["kunden"]},
}

Das ist eine Abwandlung der Beispiele aus Schritt 2 mit anderen Daten und einer Typ-Auswahl am Ende. Schreibe betroffene_berichte(graph, quelle). Sie gibt die alphabetisch sortierte Liste aller Datensätze vom Typ "bericht" zurück, die direkt oder indirekt aus quelle entstanden sind. Der Startknoten selbst gehört nie ins Ergebnis. Jeder Name steht höchstens einmal drin. Der Graph darf Zyklen enthalten, und graph darf nicht verändert werden. Ein Name, der gar nicht vorkommt, ergibt eine leere Liste.

Beispiele: betroffene_berichte(GRAPH, "shop_log") gibt ["bericht_umsatz"], betroffene_berichte(GRAPH, "crm_export") gibt ["bericht_neukunden", "bericht_umsatz"], betroffene_berichte(GRAPH, "bericht_umsatz") gibt [].

Der Graph kennt nur “entstand aus”. Welche Richtung brauchst du, und wie vermeidest du, einen Knoten zweimal zu besuchen? Was passiert mit dem Startknoten, wenn ein Zyklus zu ihm zurückführt?

def betroffene_berichte(graph, quelle):
    nach = {}
    for knoten, info in graph.items():
        for v in info["von"]:
            nach.setdefault(v, []).append(knoten)
    gesehen, offen = set(), [quelle]
    while offen:
        k = offen.pop()
        for n in nach.get(k, []):
            if n not in gesehen:
                gesehen.add(n)
                offen.append(n)
    gesehen.discard(quelle)
    return sorted(k for k in gesehen if graph[k]["typ"] == "bericht")
betroffene_berichte

Übung 2: Herkunft an jede Zeile hängen (leicht bis mittel)

Schreibe markiere(zeilen, quelle, version, zeitpunkt). zeilen ist eine Liste von dicts. Die Funktion gibt eine neue Liste zurück. Jede neue Zeile enthält alle Felder der alten und zusätzlich das Feld "_herkunft": ein dict mit genau den vier Schlüsseln quelle, version, zeitpunkt (die übergebenen Werte) und hash. Der Hash ist fingerabdruck(zeile) der ursprünglichen Zeile, also ohne _herkunft. Die Funktion fingerabdruck ist schon definiert (dieselbe wie im Text). Die übergebenen Zeilen dürfen nicht verändert werden. Eine leere Liste ergibt eine leere Liste.

Wovon wird der Hash gebildet, bevor das neue Feld existiert? Und was passiert mit der Eingabe, wenn du in das vorhandene dict schreibst?

def markiere(zeilen, quelle, version, zeitpunkt):
    return [{**z, "_herkunft": {"quelle": quelle, "version": version,
                                "zeitpunkt": zeitpunkt, "hash": fingerabdruck(z)}}
            for z in zeilen]
markiere

Übung 3: Wo ist ein Lauf nicht reproduzierbar? (mittel bis anspruchsvoll)

Ein Lauf ist ein dict mit den drei Schlüsseln eingabe (Hash der Eingabe), version (Code-Version) und ausgabe (Hash der Ausgabe), alle drei Strings. Schreibe nicht_reproduzierbar(laeufe). Sie bekommt eine Liste solcher Läufe und gibt die sortierte Liste der Paare (eingabe, version) zurück, bei denen es mindestens zwei verschiedene Ausgaben gab. Dasselbe Paar mit gleicher Ausgabe ist in Ordnung (reproduzierbar). Dieselbe Eingabe mit anderer Version und anderer Ausgabe ist ebenfalls in Ordnung (die Version hat sich geändert). Jedes Paar steht höchstens einmal drin. Die Eingabeliste darf nicht verändert werden.

Zusatzregel: Manche Läufe stammen aus einem alten Werkzeug und haben keine Version (der Schlüssel version fehlt oder sein Wert ist None). Ohne Version lässt sich ein Lauf mit keinem anderen vergleichen, also ignorierst du diese Läufe ganz: Sie erzeugen keinen Befund und stören die anderen nicht.

Beispiel: In [{"eingabe": "e1", "version": "v1", "ausgabe": "x"}, {"eingabe": "e1", "version": "v1", "ausgabe": "y"}, {"eingabe": "e1", "version": "v2", "ausgabe": "z"}] ist nur das Paar ("e1", "v1") auffällig. Kommt ein vierter Lauf {"eingabe": "e1", "ausgabe": "w"} ohne Version dazu, ändert sich nichts.

Welche Läufe vergleichst du überhaupt miteinander, und welche kommen gar nicht in Frage? Fasse die übrigen zusammen und überlege, was du je Gruppe merken musst, damit “zweimal dieselbe Ausgabe” nicht auffällt.

def nicht_reproduzierbar(laeufe):
    ausgaben = {}
    for l in laeufe:
        if l.get("version") is None:
            continue
        ausgaben.setdefault((l["eingabe"], l["version"]), set()).add(l["ausgabe"])
    return sorted(paar for paar, menge in ausgaben.items() if len(menge) > 1)
nicht_reproduzierbar

Merksatz

Lineage ist Tracing für Daten: Mit einem Graphen aus “entstand aus” suchst du bei einem falschen Wert nach oben und bei einer Änderung nach unten, und jede Zeile trägt Quelle, Version, Zeitpunkt und Hash mit, damit ein erneuter Lauf prüfbar bleibt.

Prüfstein

Eine Lieferung von Messwerten hat sich gestern im Format geändert, und im Rechnungsbericht steht heute eine Zahl, die nicht stimmt. Beschreibe, wie du mit Lineage vorgehst: Welche Abfrage (upstream oder downstream) nutzt du zuerst, welche Metadaten der betroffenen Zeile liest du, und woran erkennst du, ob ein erneuter Lauf dasselbe Ergebnis liefert?

Weiter mit Teil 2: Audit-Trail, M6-Abschluss und Rückblick.

Quelle: quellen/kursbuch-lerninhalte.md, Modul M6, Baustein “03 Lineage nachvollziehbar machen” (Warum, Kernidee mit Herkunft, OpenLineage, Merksatz, Übung). Über die Quelle hinaus (allgemeines Fachwissen, mit echtem Python 3.13 ausgeführt): Lineage als Graph mit upstream und downstream, Umkehren des Graphen, Impact-Analyse, Zyklenschutz, hashlib.sha256 über kanonisches JSON, Metadatenfeld _herkunft mit Quelle, Version, Zeitpunkt und Hash, Regel “Zeitpunkt in die Metadaten”, Reproduzierbarkeit über (Eingabe-Hash, Version, Ausgabe-Hash). Dass Pyodide hashlib.sha256 kann, aber pbkdf2_hmac und scrypt nicht, steht in den Regeln dieses Projekts (bitte bei einer neuen Pyodide-Version erneut prüfen). Dass Lineage-Werkzeuge wie OpenLineage Graphen automatisch erzeugen, ist eine Aussage der Quelle. Ob es für deinen Anwendungsfall passt, ist nicht geprüft (bitte prüfen). Alle Datensätze, Zeilen und Läufe sind erfunden.