Apache Airflow – простой и удобный batch-ориентированный инструмент для построения, планирования и мониторинга дата-пайплайнов. Ключевой его особенностью является то, что, используя Python-код и встроенные функциональные блоки, можно соединить множество различных технологий, использующихся в современном мире. Основная рабочая сущность Airflow – DAG – направленный ацикличный граф, в котором узлами являются задачи, а зависимости между задачами представлены направленными ребрами.
Те, кто использует Apache Airflow для оркестрации задач загрузки данных в хранилище, наверняка оценили гибкость, которую он предоставляет для решения шаблонных задач. Когда весь процесс разработки сводится к заполнению конфигурационного файла с описанием параметров DAGа и списком задач, которые должны выполняться. У нас в Леруа Мерлен такой подход успешно используется для создания задач по перекладыванию данных из raw-слоя в ods-слой хранилища. Поэтому было решено распространить его на задачи по заполнению витрин данных.
Основная сложность состояла в том, что единой методологии разработки витрин данных и процедур по их заполнению у нас пока нет. И каждый разработчик решал задачу, основываясь на своих личных предпочтениях и опыте. Это укладывается в один из основных корпоративных IT принципов - ”You build it – you run it”, который означает, что разработчик несет ответственность за свое решение и сам его поддерживает. Данный принцип хорош для быстрой проработки гипотез, но для однотипных вещей больше подходит стандартное решение.
Как было
Тут стоит рассказать, как велась до этого разработка для загрузки витрин данных. Разработчик пишет процедуры загрузки в GreenPlum, разрабатывает DAGи для их запуска, после чего создает по шаблону новый репозиторий на GitHub, загружает код своих DAGов и добавляет свой репозиторий в основной проект Airflow в качестве сабмодуля. При таком подходе возникали следующие трудности:
Нужно погружение в Python и Apache Airflow;
На момент начала разработки релиз основного проекта происходил раз в неделю, поэтому, чтоб увидеть свои DAGи на проде Airflow, нужно было подождать;
Основной проект постепенно разрастался и начал притормаживать при деплое;
Разбросанным по разным репозиториям кодом DAGов, выполняющих однотипные задачи, сложно управлять;
Отсутствие единого подхода также влияло и на качество SQL-кода процедур. Часто можно было встретить сложную логику по управлению параметрами загрузки, которую легко можно было «перевесить» на Airflow.
Все вышеперечисленное привело нас к мысли, что пора брать ситуацию под свой контроль и заняться разработкой стандартного решения. Анализ существующих DAG-ов показал, что большинство из них очень простые, не содержат сложных зависимостей и состоят, в основном, из DummyOperator-ов и PostgresOperator-ов. Это и послужило отправной точкой для разработки нового инструмента, который, в свою очередь, должен был:
Уметь создавать DAG-и на основе конфигурационного файла в формате YAML, в которым бы были указаны основные параметры, как-то: дата старта, расписание, параметры подключения к БД, названия запускаемых процедур, их параметры и т. д. YAML файлы должны храниться внутри корпоративного сервиса по управлению метаданными, получить их содержимое можно через API;
Быть максимально простым, иметь понятную документацию, чтобы погружение не занимало много времени;
В то же время быть максимально гибким, уметь работать с максимально возможным количеством параметров настройки DAG-а в Airflow.
Что есть
Получился примерно вот следующий шаблон для конфигурационного файла:
Из которого создается вот такой DAG:
Описание параметров
Общие параметры:
module_name – нужен для формирование DAG_ID;
pool – пул, в котором будут запущены задачи;
queue – очередь для задач;
owner – владелец DAGа;
postgres_conn_id – строка подключения к БД;
email – список емейлов для рассылки алертов;
tags – список тэгов для поиска DAGа в UI;
access_control: роль для управление DAGом;
schedule_interval – расписание для запуска DAGа;
start_date и catchup – параметры, управляющие глубиной истории загрузки. Airflow использует интервальный подход. Это означает, что временной период от start_date и до опциональной end_date (мы не используем) разбивается на интервалы, указанные в schedule_interval. Если catchup True, то запуск DAGА начнется от start_date, если False, то с текущего интервала;
schema_name – схема БД, в которой находится витрина;
task_list – список задач в DAGе.
Основные параметры задач:
task_name – соответствует task_id Airflow
task_type – тип задачи
task_schema_name - схема БД, в которой находится витрина, если схема отличается от общей
task_conn_id – строка подключения, если отличается от общей
procedure_name – процедура загрузки витрины
params – список параметров процедуры и их значений
task_depends_on – список задач, от которых зависит запуск данной задачи
priority_weight – приоритет данной задачи по отношению к другим задачам
task_concurrency - количество одновременно запущенных экземпляров задачи во всех запущенных экземплярах DAGа
Сейчас существует три типа задач(task_type):
1) Dummy – соответствует DummyOperator. Задача, которая ничего не выполняет и обычно служит начальной и конечной задачей, а также для разделения задач на блоки.
2) Обычная загрузка – соответствует PostgresOperator в Airflow
Вот так выглядит SQL-код, который генерит эта задача:
3) Множественная загрузка – много PostgresOperator(если нужно создать кучу однотипных задач, различающихся по одному параметру)
У этого типа есть свои специфические параметры:
task_multiply - может иметь 2 значения: "schema" или "params". Если указано schema", то значения из task_multiply_list добавляются в выражение SEARCH_PATH. Если "params", то значения из task_multiply_list добавляются в список параметров процедуры для параметра из списка params, у которого в значении указано 'task_multiply_list’
task_multiply_list - список значений для параметра, по которому будут создаваться однотипные задачи
В результате получается такой SQL-код.
Для “schema”:
Для “params”:
А вот так проставляются зависимости между задачами:
Куда пойдем
Внедрение инструмента позволило существенно сократить время на разработку DAGов. Глубокое погружение в Apache Airflow больше не нужно, хотя почитать про макросы и расписание все-таки придется. Шаблон конфигурационного файла заполняется минут за 10-15. Время, затрачиваемое на ревью и деплой на прод, тоже сильно сократились. Однако здесь же и кроется основная зона для развития: сейчас ревью и деплой происходят в ручном режиме. Хочется обложить все это тестами и предоставить разработчику возможность самому отправлять свои DAGи на прод.