Övervaka en datapipeline

ETL och ELT i Python

Jake Roach

Data Engineer

Övervaka en datapipeline

Datapipelines bör övervakas för dataförändringar och körningsfel

Saknad data
Ändrade datatyper
Föråldrade paket eller ändrad funktionalitet

Loggningsflöde för en datapipeline.

Logga datapipelinens prestanda

Dokumentera prestanda vid körning
Ger en utgångspunkt när en lösning misslyckas

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.

Logga varningar och fel

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.

Hantera undantag med try-except

try:
    # Execute some code here
    ...

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

Ger ett sätt att köra kod om fel uppstår

Hantera specifika undantag med try-except

Ange det specifika undantaget i except-satsen

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)

Nu kör vi en övning!

ETL och ELT i Python