Supervisión de un pipeline de datos

ETL y ELT en Python

Jake Roach

Data Engineer

Supervisión de un pipeline de datos

Hay que supervisar los pipelines por cambios en los datos y fallos de ejecución

Datos faltantes
Cambio de tipos de datos
Deprecación de paquetes o cambio de funcionalidad

Flujo de registro para un pipeline de datos.

Registro del rendimiento del pipeline de datos

Documenta el rendimiento en la ejecución
Da un punto de partida si falla una solución

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.

Registrar avisos y errores

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.

Gestionar excepciones con try-except

try:
    # Execute some code here
    ...

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

Permite ejecutar código si ocurren errores

Gestionar excepciones específicas con try-except

Pasa la excepción específica en la cláusula 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)

¡Vamos a practicar!

ETL y ELT en Python