Monitorizarea unui pipeline de date

ETL și ELT în Python

Jake Roach

Data Engineer

Monitorizarea unui pipeline de date

Pipeline-urile de date trebuie monitorizate pentru modificări ale datelor și erori de execuție

Date lipsă
Schimbări ale tipurilor de date
Deprecierea pachetelor sau schimbări de funcționalitate

Flux de jurnalizare pentru un pipeline de date.

Jurnalizarea performanței pipeline-ului de date

Documentează performanța la execuție
Oferă un punct de plecare când o soluție eșuează

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.

Jurnalizarea avertismentelor și erorilor

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.

Gestionarea excepțiilor cu try-except

try:
    # Execute some code here
    ...

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

Oferă o modalitate de a executa cod atunci când apar erori

Gestionarea excepțiilor specifice cu try-except

Specifică excepția exactă în clauza 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)

Hai să exersăm!

ETL și ELT în Python