Логирование для инженера данных: что использовать вместо print()

Привет! На связи Валерия Елпатьевская, ментор курса «Инженер данных» 👋🏻

Сегодня расскажу о важном инструменте, о котором многие почему-то забывают. О логировании.

Согласитесь, хочется понимать, что происходит после запуска кода? Представьте, вы запустили пайплайн, но данные не загрузились. Где ошибка? Что пошло не так? Хорошо, если вы хотя бы расставили print() по ходу, когда проверяли работу пайплайна. Но есть опция надёжнее.

Зачем логировать?
➖ Видеть этапы «загрузка → очистка → сохранение»;
➖ Ловить ошибки с контекстом: не просто «сбой», а «ошибка парсинга в строке 142 файла X»;
➖ Иметь историю: логи хранят время и порядок, консоль — нет.

Почему print() не подходит для продакшна?
➖ Нет времени, уровня важности и структуры;
➖ Нельзя отфильтровать или подключить к мониторингу.

Решение — модуль loguru. Улучшенный и упрощённый логер, в котором легко разобраться. Держите мини-гайд, как начать использовать его уже сейчас.

1️⃣ Установить можно через pip или conda:

pip install loguru

2️⃣ Импортировать logger и либо использовать его из коробки, либо настроить под себя:

from loguru import logger
import sys

# Настраиваем формат и теги
logger.remove()    # удалить дефолтные настройки

logger_format = (
  "{time:HH:mm:ss} | "                      # настраиваем формат даты
  "{level: <8} | "                                 # для красоты, чтобы уровни были ровно
  "{name}:{function}:{line} | "         # что хотим видеть (файл:функция:строка)
  "{extra[task_id]} - {message}"       # если хотим добавить свою информацию
)
logger.add(
    sys.stderr,                                                 # вывод в stderr
    format=logger_format,                            # добавляем формат, который выше определили
    level="INFO"                                          # с какого уровня начинаем логировать       
)

3️⃣ Использовать настроенный логер в коде:

def run_etl(file: str, task_id: str = "my_daily_task"):
    # bind() добавляет кастомный тег к этому сообщению
    logger.bind(task_id=task_id).info(f"Старт: {file}")
    try:
        # наш код...
        logger.success(f"Готово: {file}")
    except Exception as e:
        logger.error(f"Ошибка: {e}")
        raise

run_etl("orders.csv", task_id="sales_v2")

Что мы увидим в консоли:

14:22:10 | INFO          | __main__:run_etl:12 | sales_v2 - Старт: orders.csv
14:22:13 | SUCCESS  | __main__:run_etl:14 | sales_v2 - Готово: orders.csv
14:22:15 | ERROR      | __main__:run_etl:16 | sales_v2 - Ошибка: [текст ошибки]

Формат задаёт внешний вид, а extra[] и bind() — контекст. В реальных проектах теги task_id, env, user позволяют отфильтровать логи одного пайплайна среди тысяч других.

С чего начать уже сегодня:
1. Замените print() на logger.info/error/warning/debug;
2. Добавьте {time} и 1-2 тега в format;
3. Оборачивайте основные шаги, в которых возможны ошибки, в try/except с logger.error().

Логирование — это контроль и ваша страховка от стресса. Сохраняйте пост и ставьте 🔥, если было полезно!

📈 Симулейтив | 📱 ВК | 📱 YouTube | 📱 Канал о DS