Surveiller un pipeline de données

ETL et ELT en Python

Jake Roach

Data Engineer

Surveiller un pipeline de données

Les pipelines doivent être surveillés pour les changements de données et les échecs d’exécution

Données manquantes
Changement de types de données
Dépréciation de packages ou changement de fonctionnalités

Flux de journalisation pour un pipeline de données.

Journaliser les performances du pipeline

Documenter les performances à l’exécution
Point de départ en cas d’échec

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.

Journaliser avertissements et erreurs

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.

Gérer les exceptions avec try-except

try:
    # Execute some code here
    ...

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

Permet d’exécuter du code en cas d’erreur

Gérer des exceptions spécifiques avec try-except

Passer l’exception spécifique dans except

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)

Passons à la pratique !

ETL et ELT en Python