Мониторинг конвейера данных

ETL и ELT в Python

Jake Roach

Data Engineer

Мониторинг конвейера данных

Конвейеры данных нужно отслеживать на предмет изменений в данных и сбоев выполнения

Отсутствующие данные
Изменение типов данных
Устаревание пакетов или изменение функциональности

Схема логирования конвейера данных.

Логирование производительности конвейера данных

Записывайте производительность во время выполнения
Служит отправной точкой при сбое решения

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.

Логирование предупреждений и ошибок

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.

Обработка исключений с помощью try-except

try:
    # Execute some code here
    ...

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

Позволяет выполнить код в случае ошибки

Обработка конкретных исключений с помощью try-except

Укажите конкретное исключение в блоке 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)

Давайте потренируемся!

ETL и ELT в Python