Surveiller un pipeline de données

ETL et ELT en Python

Jake Roach

Data Engineer

Surveiller un pipeline de données

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

Données manquantes
Types de données qui changent
Obsolescence d'un paquet ou changement de fonctionnalité

Flux de journalisation pour un pipeline de données.

Journaliser la performance du pipeline de données

Documenter la performance à l'exécution
Offre un point de départ quand une solution échoue

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 les avertissements et les 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 si des erreurs surviennent

Gérer des exceptions précises avec try-except

Passer l'exception précise dans la clause 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