Überwachung einer Datenpipeline

ETL und ELT mit Python

Jake Roach

Data Engineer

Überwachung einer Datenpipeline

Datenpipelines sollten auf Datenänderungen und Ausführungsfehler überwacht werden

Fehlende Daten
Verschobene Datentypen
Paketveraltung oder Funktionsänderungen

Protokollierungs-Workflow für eine Datenpipeline.

Leistung der Datenpipeline protokollieren

Leistung bei Ausführung dokumentieren
Startpunkt, wenn eine Lösung fehlschlägt

import logging
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)

# Create different types of logs
logging.debug(f"Variable has value {path}")
logging.info("Data has been transformed and will now be loaded.")

DEBUG: Variable has value raw_file.csv
INFO: Data has been transformed and will now be loaded.

Warnungen und Fehler protokollieren

import logging
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)

# Create different types of logs
logging.warning("Unexpected number of rows detected.")
logging.error("{ke} arose in execution.")

WARNING: Unexpected number of rows detected.
ERROR: KeyError arose in execution.

Ausnahmen mit try-except behandeln

try:
    # Execute some code here
    ...

except:
    # Logging about failures that occured
    # Logic to execute upon exception
    ...

Ermöglicht Codeausführung bei Fehlern

Spezifische Ausnahmen mit try-except behandeln

Die spezifische Exception in except angeben

try:
    # Try to filter by price_change
    clean_stock_data = transform(raw_stock_data)
    logging.info("Successfully filtered DataFrame by 'price_change'")

except KeyError as ke:
    # Handle the error, create new column, transform
    logging.warning(f"{ke}: Cannot filter DataFrame by 'price_change'")
    raw_stock_data["price_change"] = raw_stock_data["close"] - raw_stock_data["open"]
    clean_stock_data = transform(raw_stock_data)

Lass uns üben!

ETL und ELT mit Python