Monitorowanie potoku danych

ETL i ELT w Pythonie

Jake Roach

Data Engineer

Monitorowanie potoku danych

Potoki danych należy monitorować pod kątem zmian w danych i błędów wykonania

Brakujące dane
Zmiany typów danych
Przestarzałe pakiety lub zmiany funkcjonalności

Przepływ pracy logowania dla potoku danych.

Logowanie wydajności potoku danych

Dokumentuje wydajność podczas wykonania
Stanowi punkt wyjścia przy diagnozowaniu błędów

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.

Logowanie ostrzeżeń i błędów

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.

Obsługa wyjątków za pomocą try-except

try:
    # Execute some code here
    ...

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

Umożliwia wykonanie kodu w przypadku wystąpienia błędów

Obsługa konkretnych wyjątków za pomocą try-except

Przekaż konkretny wyjątek w 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)

Czas na ćwiczenia!

ETL i ELT w Pythonie