Pipeline-Läufe planen und protokollieren

Eine Pipeline, die nur ausgeführt wird, wenn eine Entwicklerin oder ein Entwickler manuell ein Notebook öffnet, bietet über den ersten Lauf hinaus keinen geschäftlichen Nutzen. Damit sie täglich automatisch ausgeführt werden kann, muss die Pipeline als über die Kommandozeile ausführbares Python-Skript strukturiert sein: python pipeline.py. Dafür sind ein if __name__ == '__main__':-Einstiegspunkt, das Parsen von Kommandozeilenargumenten und eine korrekte Protokollierung erforderlich – die drei Säulen eines produktionsreifen Skripts.

Das integrierte Python-Modul logging ist das richtige Werkzeug für Pipeline-Protokolle – nicht print()-Anweisungen. Konfigurieren Sie mit logging.basicConfig() einen Logger sowohl für die Ausgabe auf der Konsole als auch in eine Datei. Protokollieren Sie den normalen Fortschritt auf der Ebene INFO und Fehler auf der Ebene ERROR. Dateibasierte Protokolle bleiben nach dem Beenden des Prozesses erhalten, was für die Fehlersuche bei geplanten Ausführungen, die niemand überwacht hat, unerlässlich ist.

Protokollieren Sie immer die Startzeit, die Endzeit und die verstrichene Zeit einer Pipeline-Ausführung. Dadurch entsteht eine Vergleichsbasis: Wenn die Pipeline normalerweise 45 Sekunden läuft, heute aber 8 Minuten benötigt, hat sich etwas geändert – vielleicht ist die Eingabedatei zehnmal größer oder eine Datenbankabfrage läuft langsam. Zeitgestempelte Einträge für Start und Ende machen diesen Vergleich allein anhand der Protokolldatei trivial.

Protokollieren Sie die Zeilenanzahl beim Eintritt in und beim Verlassen jedes Transformationsschritts. Ein übersichtliches Protokoll sieht beispielsweise so aus: extract: 50,000 rows → drop_nulls: 49,200 rows → filter: 47,800 rows → output: 47,800 rows. Anhand dieser Spur ist sofort ersichtlich, wie viele Zeilen in jedem Schritt entfernt wurden und ob die Zahlen erwartungsgemäß sind. Auffällige Verluste zeigen sich als Lücken zwischen den protokollierten Anzahlen.

cron ist der standardmäßige Unix-Scheduler für wiederkehrende Aufgaben. Bearbeiten Sie die Crontab mit crontab -e und fügen Sie eine Zeile hinzu, die den Ausführungszeitpunkt des Skripts festlegt. Das Format lautet: minute hour day month weekday command. Eine Pipeline, die täglich um 6:00 Uhr ausgeführt werden muss, verwendet 0 6 * * * /usr/bin/python /path/to/pipeline.py. Verwenden Sie in cron-Einträgen immer absolute Pfade, da cron in einer minimierten Umgebung ohne die PATH-Einstellungen Ihrer Shell ausgeführt wird.

Die Bibliothek schedule bietet eine auf reinem Python basierende Möglichkeit, Aufgaben in festgelegten Intervallen auszuführen, ohne cron zu verwenden. Sie ist nützlich in Umgebungen, in denen cron nicht verfügbar ist (Windows), oder wenn die Scheduler-Logik direkt im Python-Prozess enthalten sein soll. Verpacken Sie die Pipeline in eine geplante Aufgabenschleife und lassen Sie den Prozess aktiv, damit er die Aufgabe wiederholt ausführt.

Ein Pipeline-Skript sollte bei einem Fehler einen Exit-Code ungleich null zurückgeben, damit der Scheduler weiß, dass die Aufgabe fehlgeschlagen ist. Verpacken Sie die Hauptausführung in einen try/except-Block und rufen Sie bei einem Fehler sys.exit(1) auf. cron, Jenkins und Airflow prüfen den Exit-Code: Ein Wert ungleich null löst eine Warnung, eine erneute Ausführung oder eine Benachrichtigung aus. Eine unbehandelte Ausnahme, die den Exit-Code nicht setzt, kann von der automatisierten Überwachung unbemerkt bleiben.

Schreiben Sie nach einer erfolgreichen Ausführung eine kleine JSON-Zusammenfassungsdatei neben die Ausgabe. Fügen Sie den Zeitstempel der Ausführung, die Anzahl der Eingabezeilen, die Anzahl der Ausgabezeilen, die Anzahl der entfernten Zeilen und die verstrichene Zeit hinzu. Überwachungssysteme und Dashboards können diese Datei lesen, um Trends beim Zustand der Pipeline im Zeitverlauf zu verfolgen. Ein Dashboard mit den Ausgabezeilen der letzten 30 Tage macht es leicht, den Tag zu erkennen, an dem eine Datenquelle begonnen hat, weniger Datensätze zu liefern.

Wenn eine geplante Pipeline versehentlich zweimal ausgelöst wird, darf sie die Ausgabe nicht beschädigen. Gestalten Sie den Ladeschritt idempotent: Verwenden Sie einen datierten Dateinamen für die Ausgabe oder überschreiben Sie dieselbe Ausgabe mit dem neuesten Ergebnis. Verwenden Sie beim Laden in eine Datenbank if_exists='replace' oder ein UPSERT-Muster. Verwenden Sie niemals den Modus append ohne einen Schritt zur Duplikatbereinigung, da sonst jede geplante Ausführung der Ausgabetabelle doppelte Zeilen hinzufügt.

Bei Pipelines, von denen Geschäftsprozesse abhängen, ist Stille nach einem Fehler gefährlich. Richten Sie einen einfachen Alarm ein: Wenn die Datei mit der Laufzusammenfassung nicht innerhalb des erwarteten Zeitfensters aktualisiert wird, senden Sie eine E-Mail oder Slack-Nachricht. Mit Pythons smtplib können Sie bei einem Fehler eine E-Mail senden oder einen Webhook verwenden, um eine Nachricht in Slack zu veröffentlichen. Lösen Sie bei den Exit-Codes 1 oder 2 sofort einen Alarm aus, damit die Analystin oder der Analyst weiß, dass die tägliche Aktualisierung fehlgeschlagen ist, bevor es das Unternehmen bemerkt.

Führen Sie alle Bestandteile zusammen — Argumentverarbeitung, Logging-Konfiguration, Laufzusammenfassung, Fehlerbehandlung und Exit-Codes — und erstellen Sie ein vollständiges Pipeline-Skript. Dieses Skript kann in jede Umgebung übernommen, auf eine Konfigurationsdatei verwiesen und mit cron oder einem beliebigen Workflow-Orchestrator eingeplant werden. Bei jeder Ausführung erstellt es eine datierte Logdatei, eine Laufzusammenfassung und eine datierte Ausgabedatei. Dadurch ist jeder Lauf vollständig prüfbar und unabhängig reproduzierbar.

In dieser Lektion haben Sie Folgendes gelernt: eine Pipeline als Befehlszeilenskript mit Argumentverarbeitung und Logging zu strukturieren, sie mit cron einzuplanen und Fehler mithilfe von Exit-Codes ungleich null und Alarmen zu behandeln sowie Laufzusammenfassungen zu schreiben und idempotente Ladeschritte für zuverlässige automatisierte Ausführungen zu entwerfen. Herzlichen Glückwunsch zum Abschluss des Data-Analysis: Pandas und NumPy-Tracks!