Supervisión de una canalización de datos

ETL y ELT en Python

Jake Roach

Data Engineer

Supervisión de una canalización de datos

Supervisa las canalizaciones por cambios en los datos y fallos de ejecución

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

Flujo de registro para una canalización de datos.

Registro del rendimiento de la canalización de datos

Documenta el rendimiento en ejecución
Punto de partida cuando 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.

Registro de advertencias 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.

Manejo de 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

Manejo de 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