Een datapijplijn monitoren

ETL en ELT in Python

Jake Roach

Data Engineer

Een datapijplijn monitoren

Monitor datapijplijnen op dataveranderingen en uitvoerfouten

Ontbrekende data
Wijzigende datatypes
Pakketdeprecatie of functiewijziging

Logboekworkflow voor een datapijplijn.

Performance van datapijplijn loggen

Leg performance vast tijdens uitvoering
Geeft een startpunt bij een mislukte run

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.

Warnings en errors loggen

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.

Exceptions afhandelen met try-except

try:
    # Execute some code here
    ...

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

Maakt uitvoeren van code bij fouten mogelijk

Specifieke exceptions afhandelen met try-except

Zet de specifieke exception in de except-clausule

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)

Laten we oefenen!

ETL en ELT in Python