Schemas validieren und Assertions

Eine Datenpipeline verarbeitet neue Daten automatisch, oft ohne menschliche Überprüfung. Wenn sich das Schema der Eingabedatei ändert – eine Spalte umbenannt wird, sich ein Datumsformat ändert oder eine neue Kategorie hinzukommt –, sollte die Pipeline mit einer deutlichen Fehlermeldung fehlschlagen, statt unbemerkt falsche Ergebnisse zu erzeugen. Schemavalidierung mit Assertions ist der Mechanismus, der Verträge zwischen Datenproduzenten und Datenkonsumenten durchsetzt und Probleme so früh wie möglich erkennt.

Die grundlegendste Validierung besteht darin zu prüfen, ob alle erforderlichen Spalten vorhanden sind. Speichern Sie die erwartete Spaltenmenge in einer Konfiguration und stellen Sie per Assertion sicher, dass sie eine Teilmenge der tatsächlich vorhandenen Spalten ist. Diese Prüfung erkennt Umbenennungen und entfernte Spalten sofort beim Start der Pipeline, bevor nachgelagerter Code auf fehlende Spalten zugreift und tief in der Pipeline einen verwirrenden KeyError auslöst.

Validieren Sie nach den erforderlichen Spalten deren Datentypen. Eine als object geladene Datumsspalte bedeutet, dass pd.to_datetime() nicht aufgerufen wurde. Eine als float statt als int64 gespeicherte ID-Spalte weist häufig auf NaN-Werte hin, die eine Speicherung als Ganzzahl verhindert haben. Schreiben Sie für die wichtigsten Spalten Typ-Assertions wie assert df['col'].dtype == expected_type.

Schlüsselspalten wie order_id, order_date und revenue sollten niemals Nullwerte enthalten. Stellen Sie für jede dieser Spalten mit df['col'].notna().all() eine Assertion auf. Damit die Fehlermeldung unmittelbar verwertbar ist, sollten Sie die Anzahl der Nullwerte aufnehmen, sodass die für die Pipeline verantwortliche Person das Ausmaß des Problems kennt und nicht nur erfährt, dass eine Assertion fehlgeschlagen ist.

Numerische Spalten haben häufig gültige Geschäftsbereiche. Der Umsatz muss nichtnegativ sein. Die Menge muss eine positive Ganzzahl sein. Der Stückpreis muss größer als null sein. Stellen Sie diese Constraints explizit per Assertion sicher – eine durchgelassene Zeile mit negativem Umsatz würde die Summen in jeder nachgelagerten Aggregation ohne Fehlermeldung zu niedrig ausfallen lassen. Bereichsprüfungen erkennen Eingabefehler und Fehler in vorgelagerten Systemen frühzeitig.

Nach der Duplikatbereinigung sollte order_id eindeutig sein. Prüfen Sie mit assert df['order_id'].is_unique, dass diese Invariante bei jeder Ausführung der Pipeline erhalten bleibt. Verstöße gegen die Eindeutigkeit nach der Duplikatbereinigung weisen auf einen Fehler in der Bereinigungslogik oder auf eine neu eingeführte Datenquelle hin, die vor dem Zusammenführen mit dem Hauptdatensatz nicht bereinigt wurde.

Für Spalten mit einer begrenzten Menge gültiger Werte – etwa region oder category – sollten Sie per Assertion sicherstellen, dass jeder Wert in der erlaubten Menge enthalten ist. Dadurch werden unerwartete Werte erkannt, die nach einer Systemmigration oder einer Änderung der Dateneingabe im vorgelagerten System auftreten. Definieren Sie die gültigen Mengen in der Pipeline-Konfiguration, damit sie bei einer Erweiterung des Geschäfts auf neue Regionen leicht aktualisiert werden können.

Validieren Sie, dass alle Datumsangaben innerhalb des für den Datensatz erwarteten Zeitraums liegen. Eine Bestellung in der Zukunft ist unmöglich; eine Bestellung vor der Gründung des Unternehmens weist auf einen beschädigten Datensatz hin. Definieren Sie MIN_DATE und MAX_DATE in der Konfiguration und stellen Sie per Assertion sicher, dass die Spalte innerhalb dieser Grenzen liegt. Dadurch werden auch Unix-Epoch-Nullwerte (1970-01-01) erkannt, die durch fehlerhafte Zeitstempelkonvertierungen entstehen.

Eine plötzliche Änderung der Zeilenanzahl gegenüber dem vorherigen Pipelinelauf ist ein starkes Signal für ein Problem in einem vorgelagerten System. Speichern Sie die Zeilenanzahl des vorherigen Laufs in einer Konfigurationsdatei und stellen Sie per Assertion sicher, dass die neue Anzahl innerhalb eines Toleranzbereichs liegt – beispielsweise innerhalb von ±20 % der historischen Anzahl. Ein Datensatz, der zwischen zwei Läufen 50 % seiner Zeilen verliert, weist mit hoher Wahrscheinlichkeit auf einen fehlerhaften Datenexport hin.

Bündeln Sie alle Assertions in einer einzigen Funktion validate(df), die am Anfang jeder Pipeline-Phase aufgerufen werden kann. Jede Prüfung löst bei einem Fehler einen AssertionError mit einer aussagekräftigen Meldung aus. Dieses Muster macht die Pipeline selbstdokumentierend: Die Validierungsfunktion ist der maschinenlesbare Schemavertrag für den DataFrame in dieser Phase.

In Produktionspipelines ist ein harter Absturz durch assert während der Entwicklung akzeptabel, in einem geplanten Job jedoch unerwünscht. Ersetzen Sie einfache Assertions durch try/except AssertionError-Blöcke, die die Fehlermeldung protokollieren und optional vor dem Beenden einen Alarm senden. So kann das Überwachungssystem den Grund des Fehlers erfassen, statt nur einen Traceback einer unbehandelten Ausnahme zu erhalten.

In dieser Lektion haben Sie gelernt: erforderliche Spalten, Datentypen, Constraints für Nicht-Null-Werte und die Gültigkeit von Wertebereichen per Assertion zu prüfen, eindeutige Schlüssel, kategoriale Werte und Datumsbereiche zu überprüfen sowie alle Prüfungen mit Protokollierung in einer wiederverwendbaren Funktion validate() zu bündeln. Als Nächstes beschäftigen wir uns mit benutzerdefinierten Aggregationen mithilfe von apply() auf Spalten und Zeilen.