ZrobDev/etl-pipelines

Was macht etl-pipelines?

Batch-Erfassung in PostgreSQL (Standard-Datenspeicher) mit Python (uv, ruff, pytest). Lokale Transformationen: Polars 2.x oder DuckDB 2.0.

Vorgehen

1. Geben Sie den Korn an: Was bedeutet eine Zeile, durch welchen natürlichen Schlüssel wird sie identifiziert? Kein Schlüssel, kein idempotenter Ladevorgang; definieren Sie zuerst einen (oder einen Inhalts-Hash).
2. Wählen Sie das Lademuster (unten). Wählen Sie die inkrementelle Strategie. Wählen Sie den Orchestrator zuletzt.
3. Schreiben Sie den Lauf so, dass ein erneuter Durchlauf eines beliebigen Fensters denselben Endzustand ergibt. Beweisen Sie es: Führen Sie es in einem Test zweimal aus und bestätigen Sie, dass Zeilenanzahlen und eine Prüfsumme unverändert sind.
4. Fügen Sie die Observability-Prüfungen (unten) hinzu, bevor der erste Produktionslauf startet.
5. Überprüfen Sie anhand der Checkliste am Ende.

Idempotentes Lademuster

• Zeilen haben einen stabilen natürlichen Schlüssel und können sich ändern: INSERT ... ON CONFLICT (key) DO UPDATE (Postgres) oder MERGE. Fügen Sie WHERE target.hash IS DISTINCT FROM excluded.hash hinzu, damit unveränderte Zeilen nicht umgesetzt werden.
• Daten kommen als ganze Partition (Tag, Batch-ID) an und werden als Einheit neu geschrieben: Löschen-Einfügen dieser Partition in EINER Transaktion. Ein separater DELETE-Commit vom INSERT hinterlässt bei einem Absturz eine Lücke.
• Die gesamte Tabelle wird neu aufgebaut: Laden in eine Staging-Tabelle, validieren, dann austauschen (umbenennen in einer Transaktion). Leser sehen nie eine halb geladene Tabelle.
• Anhang-only Fakten ohne Schlüssel: Fügen Sie eine deterministische batch_id / Quell-Datei-Hash hinzu und lassen Sie den Ladevorgang eine bereits geladene Batch-ID ablehnen oder ersetzen.
• Niemals blind INSERT und später deduplizieren. Duplikate erreichen ein Dashboard, bevor die Deduplizierung läuft.
• Laden via COPY in eine temporäre/Staging-Tabelle, dann einen mengenbasierten Upsert. Zeilenweise Upserts aus Python sind 10-100x langsamer.

Inkrementelle Ladevorgänge

• Wasserzeichen auf einer Spalte, die die Quelle als monoton und indiziert garantiert (updated_at, Sequenz-ID). Speichern Sie es in einer Zustandstabelle, die in derselben Transaktion wie der Ladevorgang geschrieben wird, niemals in einer Datei daneben.
• Lesen Sie Wasserzeichen - Rückschau (z. B. 1 bis 24 h), nicht > Wasserzeichen. Source-Commits sind nicht nach Zeitstempel geordnet; die Rückschau plus Upsert absorbiert verspätete Zeilen. Dimensionieren Sie die Rückschau anhand gemessener Verspätung, nicht einer Schätzung.
• Wasserzeichen-Spalte unzuverlässig (wird bei Update nicht erhöht, Zeitzone gemischt, Soft-Deletes fehlen): Führen Sie einen periodischen vollständigen Abgleich durch (Schlüsselsatz-Diff oder Anzahl + Prüfsumme pro Partition), anstatt ihm zu vertrauen.
• Harte Löschungen in der Quelle sind für Wasserzeichen unsichtbar. Entscheiden Sie explizit: Abgleich durch Schlüsseldiff, oder akzeptieren und dokumentieren.
• CDC (Postgres logische Replikation, Debezium-artige Log-Leser): Verwenden Sie nur, wenn Sie Löschungen, sub-Minuten-Latenz oder wenn die Quelle nicht nach Zeitstempel abgefragt werden kann. Es fügt Slots, Retentionsrisiko (ein Leerlauf-Slot füllt die Quellfestplatte) und Ordnungs-/Snapshot-Handling hinzu. Bleiben Sie ansonsten bei Wasserzeichen.

Backfills

• Jeder Job akzeptiert --start und --end (halboffen [start, end), UTC) und leitet alles daraus ab. Kein now() innerhalb der Transformation.
• Ein Backfill ist der normale Joblauf über alte Fenster. Wenn er anderen Code benötigt, ist der normale Job nicht idempotent; beheben Sie das.
• Nach Partition aufteilen, Chunks unabhängig ausführen, Status pro Chunk aufzeichnen, sodass ein Fehler in Chunk 40 bei 40 wieder aufgenommen wird.
• Backfill in dasselbe Ziel über denselben Upsert/Partitionsersatz. Prüfen Sie auf doppelte Zählung: count(*) und count(distinct key) pro Fenster vorher und nachher.
• Drosseln Sie gegen die Quell-API/DB; ein Backfill ist ein selbst zugefügter Lastspitzen.

Wiederholungen, Zeitüberschreitungen, "Exakt einmal"

• Exakt-einmal-Zustellung gibt es nicht über ein Netzwerk. Bauen Sie Mindestens-einmal-Zustellung plus idempotente Schreibvorgänge; das Ergebnis ist effektiv-einmal.
• Wiederholen Sie nur idempotente Schritte, mit exponentiellem Backoff und Jitter, einer Obergrenze für Versuche und einer harten Zeitüberschreitung für jeden Netzwerkaufruf (connect und read getrennt setzen). Wiederholen Sie bei 429/5xx/Zeitüberschreitungen, nicht bei 4xx.
• Beachten Sie Retry-After. Wiederholungen ohne globale Ratenbegrenzung verstärken einen Ausfall.
• Ein Schritt, der E-Mails sendet, abbucht oder an einen Webhook postet, ist nicht idempotent: Schützen Sie ihn mit einem Idempotenzschlüssel, der vor dem Aufruf gespeichert wird.
• Bei endgültigem Fehler mit Exit-Code ungleich Null beenden und alarmieren; schlucken Sie niemals die Ausnahme, um einen Zeitplan grün zu halten.

Schema-Drift

• Die Landungsschicht ist permissiv (rohen Payload als JSON/Text plus Lademetadaten speichern). Die typisierte Schicht ist streng und schlägt bei Typ-/Pflichtfeldverletzungen laut fehl.
• Neue Spalten: additiv und mit Alarm erlaubt. Entfernte, umbenannte oder umtypisierte Spalten: Lauf fehlschlagen lassen und den Unterschied anzeigen; nicht auf NULL zwingen.
• Mit einem expliziten Schema validieren (Polars-Schema, Pydantic oder DuckDB DESCRIBE) an der Grenze, einmal, nicht verstreut in der Transformation.

Scrape to Warehouse

• Zwei Stufen. Stufe 1 holt und speichert die rohe Antwort unverändert und unveränderlich (Objektpfad, verschlüsselt nach Quelle, Abrufzeit, Inhalts-Hash). Stufe 2 parst Rohes in Tabellen.
• Parser-Fehler werden dann durch erneutes Parsen des gespeicherten Rohen behoben, ohne erneuten Abruf, ohne Sperrrisiko und ohne verlorene Historie.
• Zeichnen Sie URL, Abrufzeitstempel, HTTP-Status, Parser-Version für jede geparste Zeile auf. Parse ist eine reine Funktion von Rohdaten plus Parser-Version.
• Abruffehler und leere/blockierte Seiten (Captcha, 200 mit Login-Wall) werden durch Inhaltsprüfungen erkannt, nicht allein durch den Statuscode.

Observability (Minimum)

• Pro Schritt protokollieren: gelesene Zeilen, geschriebene Zeilen, abgelehnte Zeilen, Dauer, Fensteranfang/-ende, Lauf-ID. Strukturierte (JSON-)Zeilen.
• Invarianten bestätigen: Zeilen raus vs. rein innerhalb einer Toleranz; null Zeilen in einem normalerweise nicht leeren Fenster BRICHT den Lauf ab. Ein stillschweigender Null-Zeilen-Erfolg ist der häufigste unentdeckte Fehler.
• Heartbeat: Bei Erfolg und bei Start einen externen Healthcheck anpingen; Alarmieren bei verpasstem Ping (erfasst Cron, der nicht feuert, Host down). Fehler pingt ebenfalls den Fail-Endpunkt an.
• Alert bei Aktualität (maximal geladener Zeitstempel älter als Schwellenwert), nicht nur bei Job-Fehlern.
• Führen Sie eine Laufprotokolltabelle (Lauf-ID, Fenster, Status, Zählungen); sie ist auch Ihr Backfill-Hauptbuch.

Auswahl des Orchestrators

• Cron (oder systemd-Timer) + ein Skript: ein bis wenige unabhängige Jobs, keine abteilungsübergreifenden Abhängigkeiten, ein Team, das Logs lesen kann. Standard. Benötigt den obigen Heartbeat und eine Sperre (flock) gegen überlappende Läufe.
• n8n: geringes Volumen, Kleber zwischen SaaS-APIs, Webhook-getriggert, Nicht-Entwickler bearbeiten Flows. Schlecht geeignet für große Batches, schwere Transformationen oder Code-Review; schwere Logik in einem Python-Skript belassen, das es aufruft.
• Airflow 3.x: viele DAGs, abteilungsübergreifende Abhängigkeiten, erstklassige Backfills, Wiederholungen und SLAs, Audit-UI. Echte Betriebskosten (Scheduler, Metadaten-DB, Worker); nicht lohnend für eine Handvoll Jobs. Airflow 2 zu 3 ist eine Migration mit Breaking Changes; überprüfen Sie die Version, bevor Sie ein Snippet kopieren. Für dbt auf Airflow verwenden Sie Astronomer Cosmos.
• dlt: API- oder Datenbank-Erfassung, bei der Sie Schema-Inferenz, inkrementellen Zustand und Merge/Replace-Dispositionen automatisch behandeln lassen möchten. Verwenden Sie es anstatt paginierte Extraktion + Wasserzeichen + Upsert-Code von Hand zu schreiben; wenden Sie dennoch die obigen Regeln auf seine Konfiguration an.
• Container: Secrets als Compose-Secrets, gelesen aus /run/secrets/, niemals env vars oder env_file; veröffentlichte Ports an 127.0.0.1 binden oder auf einem internen Netzwerk belassen.

Anbieter-Skills

Wenn die Aufgabe dbt- oder Airflow-Spezifikationen erfordert (Modelle, Tests, DAG-Erstellung, Operatoren, Cosmos-Konfiguration), improvisieren Sie nicht: Verwenden Sie die offiziellen Anbieter-Skills, die über das de-vendors-Skill dieses Repos eingebunden werden, und greifen Sie nur zur Überprüfung des Aktualitätsstands auf die Version in references/landscape-2026-11.md zurück.

Review-Checkliste

• Schlüssel und Korn angegeben; erneuter Lauf desselben Fensters ergibt identischen Zustand (getestet).
• Löschen und Einfügen in einer Transaktion; Wasserzeichen mit dem Ladevorgang committet.
• Rückschau vorhanden; verspätete Daten und Löschungen haben eine explizite Antwort.
• Zeitüberschreitungen, begrenzte Wiederholungen mit Backoff, nicht-idempotente Schritte geschützt.
• Rohlandung unveränderlich; Parsen trennbar und wiederholbar.
• Zeilenanzahlen pro Schritt protokolliert; Null-Zeilen- und Aktualitätsalarme vorhanden; Heartbeat konfiguriert.
• Schema-Drift schlägt laut in der typisierten Schicht fehl.
• Orchestratorwahl durch Jobanzahl und Abhängigkeiten begründet, nicht durch Gewohnheit.

Konkrete SQL- und Python-Skelette: references/patterns.md. Lektionen aus der Praxis: references/gotchas.md.