Тактовый механизм управления DWH: как разгрести бесконечную очередь и не умереть28.10.2022 13:01

Привет, Хабр! Я хочу поделиться с тобой тем, как работает наш сервис по управлению загрузками данных — в общих чертах, не погружаясь супер глубоко. Основная задача сервиса — снизить трудозатраты разработчиков и сопровождения DWH на всякую рутину, связанную с управлением загрузками, в идеале — как в старой доброй кнопочке «Сделать всё хорошо».

Как работает управляющий механизм

Итак, речь пойдет об автоматизированной системе под названием «Модуль управления» или просто УМ.

Модуль управляет бОльшей частью процессов в Централизованном хранилище данных (ЦХД). Большей — потому что есть обходные пути для особенных источников и витрин. Всегда же должен быть план Б =).

В ЦХД стекаются данные из информационных систем, которые используются в компании по всей стране: в цифрах это 300+ систем-источников и примерно 20 тысяч таблиц-источников, занимающих 650 ТБ, распределенных между Greenplum и Hive, а также широкий слой детальных данных и множество витрин.

Процессы, которыми управляет УМ — это примерно то же, что и у всех: поэтапные загрузки данных в различные слои DWH, сбор статистики, онлайн-проверка качества данных, рассылка оповещений и прочая стандартная дребедень.

Базовые принципы

1. Максимальная автоматизация всех процессов

Первый и главный принцип, который заложен в саму суть системы — «автоматизируй всё, что можно автоматизировать». Мы оставляем людям минимальную ручную настройку и разбор ошибок. Все процессы завязаны на метаданные: когда есть полное описание таблицы, можно и DDL сгенерировать, и DML автоматически во время загрузки сформировать, и алгоритм обновления выбрать без участия человека. Про всякие попутные логирования и говорить нечего.

Специально обученный человек один раз разрабатывает несколько унифицированных шаблонов загрузки данных и прописывает в системе логику выбора шаблона на основе метаданных (например, один шаблон для таблиц, которые должны храниться в Greenplum, другой — для тех, кто хранится в Hive). И всё — новые сущности могут грузиться по этому шаблону без дополнительных приседаний.

2. Все процессы разбиты на атомарные шаги

Загрузки данных, проверки КД, синхронизация метаданных, сбор статистики и т.д. — все процессы, которые умеет контролировать УМ, зарегистрированы в репозитории и описаны как набор параметризированных операций-шагов, между которыми есть направленные связи (как в ориентированном графе). Каждая операция должна быть неделимой (атомарной) и перенакатываемой (идемпотентной).

3. Процесс — это направленный ациклический граф

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

Например, процесс загрузки данных в слой оперативных данных состоит из четырех этапов:

1. Загрузка в промежуточную таблицу (aka staging), очищаемую перед каждой загрузкой;

2. Проверка качества загруженных в staging данных (на дубли, на допустимые значения и пр);

3. Актуализация целевой таблицы постоянного хранения (aka target) одним из стандартных алгоритмов;

4. Проверка качества загруженных в target данных (бизнесовые проверки).

Этот процесс универсальный, не привязанный к конкретной таблице источника или таргету — наоборот, target задается параметром при старте процесса, а все остальное вычисляется из метаданных.

В отличие от загрузок ODS, расчеты таблиц детального слоя и витрин часто объединены в группы по предметным областям, и каждому «кубику» заранее задана конкретная таблица ХД. Представление таких процессов выглядит чуть более сложно, например, так:

Но при сборке такого процесса все равно используются универсальные кубики, для каждого из которых надо только конкретную таблицу задать.

Хранится описание процессов в трех настроечных таблицах: Таблица процессов, Таблица шагов процесса и Таблица связей между шагами.

4. Регистрация задачи вместо немедленного выполнения

Когда через API прилетает очередное задание на запуск процесса, УМ вычисляет все необходимые параметры операций-шагов процесса, формирует команды и добавляет операции в общую очередь операций. Непосредственные расчеты начнутся позже — когда все пререквизиты отработают и получат успешный статус. Пререквизитами операции являются как операции того же процесса, которые выше по графу, так и другие процессы из общей очереди, использующие тот же объект ХД.

В целом, порядок выполнения операций определяется принципом FIFO: новые загрузки встают в конец очереди, а оркестратор отбирает из очереди первые N операций, чтобы отправить на выполнение. Дополнительно реализовано множество способов управления очередью и процессами в ней:

• отмена — операции и процесс удаляются из очереди, освобождая дорогу другим;

• пауза — операции не отправляются на выполнение, но остаются в очереди (и блокируют зависимые процессы);

• рестарт — процесс начинается заново, как будто до этого ничего не выполнялось;

• рекавер — для процессов, упавших в ошибку: продолжить с места падения, повторить ошибочную операцию;

• игнор ошибки операции — для продвинутых пользователей с расширенными правами: продолжить с места падения без повтора ошибочной операции;

• подвинуть процесс вверх/вниз в очереди — на примере с картинки можно поменять местами процессы B и C, т.к. они еще не начали выполнение;

• подвинуть операцию без пререквизитов вверх/вниз очереди — операций без пререквизитов может быть много (в пиковые нагрузки на сервис в работе может быть несколько тысяч процессов). Такие операции образуют подочередь (очередь свободных операций), и обычно сортируются по дате добавления, но на этот порядок также можно влиять.

Надо отметить, что на порядок выполнения операций внутри одного процесса влиять нельзя — т.к. это может привести к ошибкам и дефектам данных.

5. Дискретно-непрерывный подход к ведению загрузок

Модуль управления — это механизм, работающий по принципу «включился — поработал — выключился — повторил через N минут». Одну итерацию такого цикла мы называем тактом. Частота тактов настраивается в зависимости от контура (среды) — на продуктиве это раз в минуту.

В начале каждого такта механизм определяет перечень операций, которые можно выполнить немедленно, асинхронно запускает их и завершает такт. Момент с асинхронностью здесь очень важен — система должна уметь запускать операции без ожидания завершения, т.к. самые тяжелые работают по несколько часов. Операция выполняется в обертке — стандартном джобе, который логирует и журналирует ход выполнения.

«Перечень операций, которые можно выполнить немедленно» — это как раз те операции из очереди, у которых не осталось пререквизитов и которые попадают в число «первых N».

Также в начале такта УМ выполняет управляющие действия над процессами — полностью выполненные закрывает, другие инициализирует.

6. Остановка ветки в случае ошибки

Когда всё идет гладко — загрузки грузятся, шестеренки крутятся, пользователи занимаются своими делами. Но как только происходит ошибка выполнения — блокируется всё, что по пайплайну зависит от упавшего этапа.

Игнорировать ошибку и продолжить процесс загрузки обычно нельзя, т.к. невыполнение одного этапа отрицательно скажется на результате остальных. Отменить весь процесс, в котором произошла ошибка, тоже нельзя — можно упустить массовую проблему или получить дырку в данных.

Из этого вытекает следующий факт — упавший процесс остается висеть в очереди. Когда зарегистрировано несколько загрузок по одному и тому же объекту, то ошибка одной загрузки блокирует не только pipeline-ветку процесса (например, какая-то витрина не стартанет, пока не рассчитается таблица DDS), но и остальные загрузки в тот же объект (т.е. ошибка при расчете таблицы за 1 января блокирует загрузки той же таблицы за другие даты). При этом на несвязанные загрузки влияния вообще нет и не должно быть — работают себе независимо и в ус не дуют.

В результате получается два момента, которые стоит отметить:

• с любой проблемой, даже просто обрывом коннекта, должен разбираться ответственный сотрудник;

• если ошибкой никто не занимается, то по объекту может накопиться приличный хвост, который начнет рассасываться только после решения проблемы. 

Второй момент кажется минусом –, но это только на первый взгляд. На самом деле это своеобразная защита от проблем в данных, которая позволяет разбираться с ошибками в порядке критичности, не бросаясь сразу на все и не переживая, что от задержки что-то испортится или сломается.

Кстати, в бэклоге находится задача по разделению ошибок коннекта и остальных проблем. Предлагается при ошибке коннекта к БД повторять попытку выполнения операции через некоторое (настраиваемое) время. Ну и ограничить число попыток, конечно, чтобы в бесконечный цикл не уйти.

7. Защита открытых транзакций и логическая блокировка объектов

Статусы операций живут в специальном «журнале операций» — таблице, в которую попадают записи о задействованных в загрузках объектах ХД и операциях над ними. Перед началом каждой операции в журнал вставляется строка с указанием на объект, типом операции и статусом Running. После выполнения операции статус записи в журнале меняется на Success или Error в зависимости от результата. Записи в журнале живут, пока не завершится вся загрузка (все операции процесса).

В результате журнал не только отражает ход работы процесса, но и создает логическую блокировку объекта на весь период загрузки — даже после успешного завершения атомарной операции аналогичную новую операцию начать нельзя, т.к. следующие операции того же процесса могут использовать данные заблокированного объекта. Например, данные staging-таблицы нельзя перетирать, пока они успешно не зальются в target.

Логическая блокировка объекта в журнале реализована просто как уникальный индекс над таблицей-журналом.

Это невероятно полезная штука, супер-резерв, который несколько раз спасал нас от проблем потери данных. Не часто, но было такое, что что-то где-то задублилось, сломалось, неверно настроилось — и УМ пытался запустить одну и ту же операцию дважды. А т.к. вставка записи в журнал происходит перед выполнением операции, то такой номер не проходит — кто первый запись в журнал вставил, тот таблицу и танцует загружает. Уникальный индекс в этом плане надежен как скала =).

Технологический стек

Далее в статье упоминаются продукты таких вендоров как IBM и Oracle. Я понимаю, что в свете событий 2022 года использование их в проекте может быть, мягко говоря, затруднено или неразумно. Но пока так исторически сложилось. Принципы и подходы, описанные в статье, универсальны — ровно ту же историю можно реализовать на опенсорсных продуктах (и нам тоже придется постепенно перейти).

Система реализована как связка Oracle БД + ПО-оркестратор + рутины.

Упрощенная схема взаимодействия компонентов

Рутины (именованный исполняемый код) реализуют логику обработки данных — бизнесовую или техническую. Они могут быть реализованы как угодно: exe-файл, python-скрипт, bash-скрипт, процедура/функция БД, объект ETL-инструмента и пр. Всякие исполняемые файлы могут запускаться в контейнере, иметь собственные настройки/конфиги/окружение и пр. Все это неважно для УМ — главное, чтобы рутина была описана в настройках УМ как исполняемая параметризованная команда.

Пример: рутина актуализации ODS (т.е. обновления target-таблицы данными staging) у нас реализована как python-скрипт. Параметризованная команда вызова этого скрипта выглядит примерно так:

$python_path $py_act_folder/act_ods.py -t $P_OBJECT_NAME -src $P_OBJECT_SRC_CODE 
-st $P_START_DTTM -end $P_END_DTTM -upload $P_UPLOAD_DTTM 
-env $P_ENV -pkg_id $P_PKG_ID -init $P_INIT

Все, что начинается с $, является либо параметром системы, либо параметром рутины. Параметры рассчитываются в момент регистрации загрузки, и в очередь попадает уже полностью готовая команда.

Добавить в УМ новую рутину легко — в настроечных таблицах описывается специальный объект: сама команда, способ выполнения (т.е. как выполнять — как команду ОС или, например, в БД отправлять) и формулы для расчета параметров. Далее указанный объект можно встраивать в любой процесс.

Oracle БД используется не только как хранилище метаданных, логов, журналов и настроек, но и как основной мозг системы (backend). Вся логика управления процессами зашита в хранимые объекты БД — представления и PL/SQL-пакеты.

Oracle как основа бэкенда был выбран сознательно — дополнительная прослойка между БД и оркестратором усложнила бы систему, но при этом не принесла бы особого профита. Реализовывать внутреннюю логику в оркестраторе мы пробовали, и нам не понравилось — очень тяжело поддерживать по сравнению с текущим вариантом. Код, написанный на SQL и PL/SQL, гораздо проще версионировать, накатывать изменения, оптимизировать и отлаживать, можно включить в CI/CD и пр. Также дополнительный плюс — нет особой привязки к оркестратору — переехать на другое ПО в случае чего не такая уж и проблема. Сменить СУБД, конечно, будет несколько сложнее.

Основные объекты в БД:

1. Журнал операций;

2. Очередь операций;

3. Таблица-пул, в которую кладется перечень операций, отобранных из очереди для выполнения в текущем такте. Каждой операции сопоставлено имя объекта оркестратора, который будет её выполнять;

4. Процедура управления, отвечающая за всю внутреннюю кухню обработки процессов и заполнение пула;

Кстати, таблица-пул появилась не сразу — поначалу оркестратор вычитывал список операций непосредственно из очереди. Но запрос к очереди довольно тяжелый, и оказалось, что такой подход иногда приводит к накладкам.

Оркестратор отвечает за «тактовость» работы механизма и «дергает за ручку» API остальных объектов системы. Он может быть реализован в любом инструменте, позволяющем обращаться к базам разных СУБД, запускаться по расписанию и отправлять команды в операционную систему. ETL-инструменты вроде Informatica, Airflow и даже NiFi вполне подходят. Мы выбрали IBM Datastage — на момент выбора у нас были лицензии, поддержка IBM и собственная экспертиза, плюс на стадии пилота Datastage оказался лучше, чем Airflow — в нем меньше багов и отказоустойчивость выше.

В оркестраторе реализовано несколько объектов, описывающих основные процессы — один головной, отвечающий за общее руководство, и дочерние, реализующие выполнение одной операции. Дочерних объектов столько, сколько реализовано способов выполнения операций — в нашем случае два: команды ОС и вызов процедуры БД. Когда появится новый способ выполнения, придется доработать оркестратор: создать новый объект и прописать его в настроечных таблицах БД, чтобы правильно заполнялась таблица-пул.

Процессы оркестратора: основной и дочерний

Фишки и плюшки

1. Балансировка нагрузки на БД

Количество одновременно работающих в ХД процессов надо контролировать — если отправить на выполнение одновременно 100500 DML-операций, база может крякнуть и упасть. На продуктивном контуре под загрузки ХД выделены ресурсы, позволяющие работу 300 запросов одновременно (это на технические нужды, пользователи в другой группе тусят), но иногда и 300 слотов мало — в очереди может быть несколько тысяч незаблокированных операций. Такие аппетиты ограничиваются так: каждый такт УМ выбирает из очереди первые N операций, где N = Limit — Count_R.

Limit — максимально допустимое число одновременно выполняемых DML-операций, при котором гарантированно не падает БД.

Count_R — число операций, работающих в данный момент.

Limit пока задается пользователем вручную как глобальный параметр системы, но в бэклоге есть задача «Автоматическая адаптация лимита», в которой предлагается периодически проверять состояние Greenplum и повышать/понижать Limit.

2. Пока нет проблем масштабирования

Количество систем-источников постоянно растет, но это не сказывается на производительности. Фактически, мы упираемся в возможности БД ЦХД (Greenplum) в части числа одновременных запросов и свободное место. Также ПО, на котором работает оркестратор-запускалка, должно поддерживать одновременную работу как минимум Limit процессов/джобов. УМ можно развернуть на разных ПО — в данный момент у нас IBM Datastage, ресурсов и отказоустойчивости которого пока хватает. При этом у самого УМ, отдельно от ПО и от Greenplum, ограничений практически нет.

3. Все регламенты дружат и шарят ресурсы

Благодаря очереди и круглосуточной работе загрузки в ХД могут идти 24/7/365 — по разным объектам, за разные даты и периоды. Нет такого, что сейчас мы грузим все за 1 июня, а вчера грузили 31 мая. Одновременно могут идти процессы инициализирующих и регулярных загрузок с абсолютно разными периодами. Загрузка одного и того же объекта может быть включена в очередь несколько раз за разные периоды. Окончание загрузки одного объекта может стать триггером для старта загрузки другого.

Реальный пример: в последнюю версию скрипта, обеспечивающего обновление целевой таблицы ODS, закрался некий баг, который немного портил данные. Баг волшебным образом миновал все тесты и выкатился на прод, где прожил несколько дней, пока его не обнаружили и не исправили. В результате пришлось загружать данные источника в ODS заново за несколько дней. Сотрудник, разбиравший проблему, после исправления скрипта просто выполнил вручную запуск загрузки несколько раз за нужные периоды — без ожиданий, когда добежит одно, чтобы запустить другое. Загрузки зарегистрировались, встали в очередь и последовательно, в хронологическом порядке сами отработали без дополнительных танцев с бубном. Таблица детального слоя, при расчете которой используется описанная таблица ODS, стартовала сама* за эти же периоды, согласно своим настройкам запуска. От таблицы DDS веером разошлись загрузки витрин, опять же абсолютно самостоятельно, без дополнительных пинков.

*За автоматический запуск расчета по событию или группе событий отвечает другая система (сервис подписок), о которой я могу рассказать в отдельной статье, буде у сообщества возникнет такой интерес.

У внимательного читателя мог возникнуть вопрос –, а зачем запускать несколько загрузок за разные периоды, если можно запустить одну за один общий большой период?

4. Все операции порождают событие

Каждая загрузка, проходя очередной этап (выполнив операцию), создает событие, регистрируемое в системе. Эти события используются как триггер для запуска следующих по пайплайну загрузок в слой детальных данных и витрины. Также события используются в различных метриках и мониторинге выполнения регламента — отслеживается выполнение плана.

Выхлоп

За месяц УМ обрабатывает суммарно более 300 тысяч разноплановых процессов и полтора миллиона операций. Их поддержкой занимается команда из 10–12 человек. Все тщательно залогировано, и логи хранятся 3 года. По любому процессу можно поднять историю и, например, провести анализ изменения скорости выполнения тяжелой операции.

Примерно 85% загрузок из источников в ODS реализуются без участия разработчика, на готовых процессах (если источник более-менее стандартный и не требует специальной реализации).

Разработка загрузок в DDS и витрины значительно ускорена — разработчики тратят на создание нового процесса и встройку его в общую систему считанные минуты, а основную часть времени — на реализацию и отладку бизнес-логики. Правда, на погружение в работу с УМ тратится больше времени, это да.

Команда сопровождения избавлена от многих проблем, типичных для хранилищ, реализованных в лоб — таких как управление количеством одновременных процессов, порядок выполнения равнозначимых процессов, необходимость ручного контроля выполнения и последовательного запуска зависимых процессов в случае накладок или сбоев и т.д.

Очень большой плюс — круглосуточная работа без дежурных. Наш кластер Greenplum не позволил бы провести, например, все дневные загрузки за одну ночь или за один рабочий день, а УМ выполняет большую часть работы по контролю и запуску сам, что позволяет не сильно страдать при отставании регламента и нагнать его в любой момент.

На логах и журналах УМ собирается самая разная статистика — выполнение регламента, число ошибок в час, среднее время выполнения и подозрительные зависания операций и прочее. Также на них натравлена Grafana — на графиках в режиме реального времени можно наблюдать за показателями системы и получать от Telegram-бота оповещения при появлении проблем.

© Habrahabr.ru