Задание представляет собой апдейт задания 2 Было необходимо переписать ДАГ с добавлением ветвления.
Получилась следующая структура:
Скрипты разбиты логически по файлам:
-
data_collector_hw4_dag.py - собственно сам ДАГ и пременные-операторы
-
data_collector_hw4_operators.py содержит операторы:
LoadOrdersOperator- загрузка данных о заказах;LoadTransactionsOperator- загрузка данных о транзакциях;LoadGoodsCustomersOperator- загрузка данных о товарах и покупателях из PostgreSQL.
все загрузчики чистят дубли, надеюсь...
FinalSaveDataOperator- сохранение в PostgreSQL;TelegramErrorSendOperator- отправка ошибки в телеграм ботом. Для корректной работы необходимо наличие .env файла, содержащего себе
TELEGRAM_PROXY=socks5://login:pass@host (использование необязательно)
TELEGRAM_BOT_TOKEN='токен телеграма'
TELEGRAM_CHAT_ID='чат телеграма'
- data_collector_hw4_check_operators.py содержит операторы:
PostgreCheckOperator- проверка соединения с БД, срабатывает, когда нет соединения, не подходят логин/пароль, отсутствуют прописанные соединения в Airflow;DataCheckOperator- проверка и сбор финального дата сета. Выкидывает, когда на выходе получается записей меньше 70% от таблицы транзакций или в целом меньше 10 записей. Запись принимается некорректной, когда- ФИО после чистки содержит меньше 2х слов
- Отстутсвует название товара
- Количество товаров меньше 1
- data_collector_hw4_utils.py содержат дополнительные функции сохранения/чтения данных
Для задания был добавлен класс TelegramEventer, который в зависимости от состояния "таски" отправляет в телеграм сообщение о состоянии. Уведомления приходят при корректном завершении таска "all_success", а так же при перезапуске и неудачах выполнения.
Так же был добавлен ДАГ hw_5_canary_test, который рандомно собирает ya.ru с вероятностью ошибки приверно 33% для проверки работоспособности отправки состояний в телеграм. Для дага прописаны коллбэки на статусы и sla_miss_callback. О состояних пишет в телеграм.
На сервере подняты Grafana и Prometheus.
Код написан в образовательных целях на онлайн-курсе Airflow 101.