Monitorování datového potrubí

ETL a ELT v Pythonu

Jake Roach

Data Engineer

Monitorování datového potrubí

Datová potrubí je třeba monitorovat kvůli změnám dat a chybám při provádění

Chybějící data
Změny datových typů
Zastaralé balíčky nebo změny funkcionality

Schéma logování pro datové potrubí.

Logování výkonu datového potrubí

Zaznamenává výkon při provádění
Poskytuje výchozí bod při selhání řešení

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.

Logování varování a chyb

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.

Ošetření výjimek pomocí try-except

try:
    # Execute some code here
    ...

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

Umožňuje spustit kód v případě výskytu chyb

Ošetření konkrétních výjimek pomocí try-except

Předejte konkrétní výjimku v klauzuli 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)

Vyzkoušejte si to!

ETL a ELT v Pythonu