Ещё один день из жизни Sr Data Engineer.

Ускорил таску пайплайна с 1..7 часов до 0.5-3 минут за счёт уменьшения обращений к базе.

Иногда сложно угадать, какой популярностью начнёт пользоваться самописный сервис. И тогда понимаешь, что не просто так на собесах просят оценить асимптотическую сложность алгоритмов в O(n). Даже линейный бывает медленнее, чем нужно.

Сегодня речь пойдёт о сервисе парсинга логов, который унифицирует подключение аналитики студий-партнёров к единому хранилищу данных. Преобразование сырых данных в Parquet’ы и (пере)создание расширяемых вьюшек (представлений, view) в разных разрезах вынесены в Airflow с PyArrow и PySpark. Хранилище - HDFS с Hive’ом, структуры вьюшек хранятся в Redis’е и загружаются в python-объект.

Проблема была в том, что даги Airflow проверяли наличие очередной вьюшки прямолинейной попыткой CREATE OR ALTER. То есть обращений к базе было столько же, сколько вьюшек в принципе, и большую часть времени занимала работа вхолостую. Таких вьюшек за пару лет накопилось полторы тысячи, и даг каждый час пытался ходить полторы тысячи раз в продовую базу и зависать на каждой вьюшке секунд на 15-30.

Что я сделал — нашёл способ обратиться к Metastore Hive’а напрямую из дага, написал SQL-запрос для определения дата-времени последнего изменения DDL-структуры с максимальной фильтрацией только нужных вьюшек через регулярное выражение. Вывел значения в словарь вида {“схема.таблица”: “timestamp”, …}. При каждом реальном изменении структуры вьюшки добавил запись в объект и затем в редис поле с таймстемпом внесения изменений utc_updated_dttm = datetime.now(timezone.utc). Соединил обе части простой условной проверкой: “если есть изменения структуры в объекте и они новее, чем последнее обновление вьюшки, запускай процесс”. Добавил также обработку текущих объектов, где поля utc_updated_dttm ещё нет.

Результат — время холостой проверки, что ничего не изменилось, упало до 30..150 секунд, в зависимости от нагрузки на Hive (один запрос для выгрузки метаданных пачкой и проверка условий в памяти на воркере). До изменений занимало 40..400 минут (n последовательных запросов к Hive, по одному для очередной вьюшки).

С какими проблемами столкнулся:
1. Интеграция задействует много сервисов — HDFS, Hive server, Hive Metastore, Redis, Airflow. Тестового контура нет, в контейнерах поднимать для одной задачи избыточно. Тестировал “грязно” в python-файле с копипастом всех классов и методов, закомментировав все “пишущие” операции. Выводил в STDOUT.
2. Долго дружил таймстемп, прочитанный из Redis’a (строка на основе timezone aware datetime) и таймстемп, прочитанный из Hive (строка из timezone naive datetime), несмотря на UTC в обоих случаях. Пришлось привести строки к datetime объектам через strptime и потом — к epoch time через .timestamp(). Возможно, в будущем перепишу элегантнее через запись в редис в том же формате, в котором читает из хайва.
3. Для отслеживания того, что ложно-отрицательных срабатываний нет (т.е. что изменения пропускаются так же, как их отсутствие), настроил дашборд в Grafana, немного допилив выгрузку в Victoria Metrics. Сами логи для определения конкретной вьюшки с ожидаемыми изменениями смотрел напрямую в докер-контейнере с развёрнутым на проде сервисом.
4. Не один час заняло определение мест, в которые нужно внести изменение. Сказался небольшой опыт поддержки ООП приложений, состоящих из нескольких сервисов. Просматривал глазами определение функций и классов, которые использовались в итоговой таске, раскручивал до базовых объектов.

#data #de #data_engineer

Вступай в ОМ | Расти в IT | Присылай историю