• что за зверь этот Airflow, из каких компонентов состоит и как они между собой взаимодействуют
• про основные сущности Airflow: пайплайны, которые называются DAG, Operator и еще про несколько вещей
• как преуспеть в разработке на Airflow
• как мы внедрили генерацию пайплайнов и так называемое «декларативное писание пайплайнов»
• про плюсы и минусы использования Airflow

Что такое Airflow

Архитектура Airflow

Компоненты Airflow

• Первый нюанс заключается в том, что в один момент времени может работать только один инстанс Scheduler’a. Это значит, что на текущий момент невозможен режим работы в High Availability (разработчики планируют добавить Scheduler HA в Airflow версии 2.0).
• Второй нюанс: в определённый момент на запуск могут отправиться несколько пайплайнов, из-за чего фактический старт задач может откладываться на несколько минут. Бывалые дяденьки рассказывают, что запаздывания могут затягиваться на полчаса-час или больше, но лично я с таким не сталкивался.

• Первый, самый простенький – это SequentialExecutor. Он последовательно запускает прилетающие задачи, а на время их выполнения приостанавливает шедулер.
• LocalExecutor на каждую прилетающую задачу стартует новый процесс, с ним появляется возможность запускать несколько задач в параллель, поэтому LocalExecutor чуточку лучше предыдущего экзекьютора. Есть один нюанс: если в качестве базы метаданных используется что-то типа однопоточного SQLite, ваш LocalExecutor превращается в SequentialExecutor.
• CeleryExecutor позволяет иметь несколько воркеров, работающих на разных машинах. Celery – это распределенная очередь задач, которая под капотом использует RabbitMQ или Redis. При запуске воркера ему можно указать названия очередей, из которых он будет принимать задачи от шедулера.
• DaskExecutor запускает задачу с помощью Dask – библиотеки для параллельных вычислений.
• KubernetesExecutor на каждую задачу запускает новый pod в Kubernetes.
• DebugExecutor создан для запуска и отладки пайплайнов из IDE.

Сущности Apache Airflow

dag = DAG(
   dag_id="load_some_data",
   schedule_interval="0 1 * * *",
   default_args={
       "start_date": datetime(2020, 4, 20),
       "owner": "DE",
       "depends_on_past": False,
       "sla": timedelta(minutes=45),
       "email": "<your_email_here>",
       "email_on_failure": True,
       "retries": 2,
       "retry_delay": timedelta(minutes=5)
   }
)

Очень важный момент: поскольку Airflow разрабатывался международной компанией, он работает только по UTC. На текущий момент нет вменяемого способа заставить Airflow работать в другом часовом поясе, поэтому нужно постоянно помнить про разницу нашего часового пояса с UTC. В версии 1.10.10 появилась возможность менять таймзону в UI, однако это касается только веб-интерфейса, пайплайны все равно будут запускаться по UTC.

Оператор

run_sql = HiveOperator(
   dag=dag,
   task_id="run_sql",
   hive_cli_conn_id="hive",
   hql="""
       INSERT OVERWRITE TABLE some_table
       SELECT * FROM other_table t1
       JOIN another_table t2 on ...
       WHERE other_table.dt = '{{ ds }}'
   """
)

notify = SlackAPIPostOperator(
   dag=dag,
   task_id="notify_slack",
   slack_conn_id="slack",
   token=token,
   channel="airflow_alerts",
   text="Guys, I'm done for {{ ds }}"
)

run_sql >> notify

Идемпотентность

Самые распространенные операторы Airflow

• BashOpetator и PythonOperator. С ними все понятно: они отправляют на выполнение bash команду и python функцию соответственно.
• Есть огромное множество операторов для отправки запросов в различные базы данных. Поддерживаются стандартные Postgres, MySQL, Oracle, Hive, Presto. Если оператора для вашей любимой базы данных почему-то нет, можно использовать более общий JdbcOperator или написать свой, Airflow это позволяет.
• Sensor – это оператор, который при запуске проверяет выполнение определенного условия. И если оно не выполняется, оператор на какое-то время засыпает. Например, оператор может проверять количество строк в таблице, наличие файлов в файловой системе. Есть сенсор, которому можно сказать: подожди до 3 часов ночи, после этого передай эстафету следующей задаче. Сенсоры легко кастомизируются под задачи. Например, мы используем их для проверки готовности данных во внешних системах, чтобы не запускать тяжелый отчет на неполных данных.
• BranchPythonOperator – это оператор ветвления, который на основании определенного условия, заданного python кодом, решает, какую задачу надо запустить следующей.
• DockerOpetator запускает Docker-контейнер на воркере. Тут нужно понимать, что внутри Docker-контейнера может запуститься все, что угодно. Поэтому очень важно при этом мониторить ресурсы воркера, чтобы они внезапно не закончились.
• KubernetesPodOperator запускает новый pod в Kubernetes.
• DummyOperator выполняет роль пустышки и создан для того, чтобы склеивать различные участки пайплайна между собой.

Какие операторы мы используем в Lamoda

• Наша основная рабочая лошадка – это LamodaDockerOperator. Как понятно из названия, мы немножко подпилили напильником стандартный оператор: добавили монтирование актуальных конфиг-файлов Hadoop, прокидывание некоторых переменных окружения по умолчанию и еще парочку мелочей. В основном мы запускаем с помощью LamodaDockerOperator Spark-джобы в кластере, либо код на python.
• LamodaHiveОperator – тоже оператор, который мы изменили. Он отправляет запросы в Hive. Мы заметили, что у стандартного оператора в некоторых случаях запрос почему-то завершается ошибкой, при этом оператор говорит нам, что запрос завершился успешно. Мы заменили тип хука, который использовался под капотом у этого оператора, с HiveCliHook на HiveServer2Hook, и все стало замечательно.
• Следующий интересный оператор – это ExternalTaskSensor. С помощью него мы связываем пайплайны между собой. У нас есть набор базовых пайплайнов, которые загружают в Hadoop данные из внешних источников. Другие пайплайны, которые используют эти данные, могут подписаться на завершение либо всего базового пайплайна, либо какой-то конкретной его задачи, и начать выполняться только после того, как завершатся базовые. Раньше у нас был самописный механизм, который работал через файлы-флаги на HDFS, но буквально несколько недель назад мы переехали на встроенный механизм Airflow.
• BashOperator, PythonOperator – очевидно, запускают bash-команду и python код соответственно.
• И есть несколько небольших сенсоров и операторов, которые мы написали под свои нужды. Они очень-очень маленькие, я решил их сюда не выносить.

Другие полезности Airflow

• Variables – это переменные, смысл которых, думаю, очевиден. Здесь мы храним вещи, которые нужны для запуска пайплайнов. Например, там лежат названия популярных схем в Hive, пути к часто используемым директориям в HDFS, идентификатор окружения и прочее. У нас различаются значение для некоторых переменных в dev-среде и prod-среде, таким образом мы изолируем ландшафты друг от друга.
• Connections – это сущность, в которой хранится информация о подключении к внешним источникам. В Airflow поддерживается множество источников: от банальных http и ftp, заканчивая популярными базами данных и облачными провайдерами.
• Hooks – это интерфейсы для внешних источников и баз данных, часто выступающие ключевым звеном для работы операторов и сенсоров.
• SLA и тайм-ауты. При создании задачи можно указать, сколько у нее есть времени на выполнение. Если указать SLA и задача не успеет выполниться, то на почту придет письмо, что такой-то негодяй почему-то не успел вовремя. Про тайм-аут понятно: если этот негодяй совсем-совсем не успел, то Airflow его безжалостно прикончит.
• Следующая интересная штука – это XCom, что расшифровывается как cross-communication. Он создан для передачи данных между задачами: это могут быть название файла, строчка из базы данных или json-объект. Максимальный вес – до 48 килобайт.
• Есть некоторое количество графиков – например, диаграмма Ганта. Здесь интересно проследить ту самую задержку перед запуском, которую я упоминал выше. Даже несмотря на то, что здесь выставлен параллелизм 5, мы ожидаем, что задачи стартанут одновременно, но нет, лаг на запуск присутствует.

Как преуспеть в разработке на Airflow

• Полезно держать каждый пайплайн (или генератор пайплайнов, об этом ниже) в отдельном файле. Я сразу знаю, в какой файл нужно залезть, чтобы посмотреть на нужный пайплайн или генератор.
• Хорошей практикой является подход, когда пайплайн состоит из задач, которые максимально атомарны и идемпотентны. Конечно, можно создать одну задачу со здоровенным баш-скриптом, делающим все на свете. Но при разделенных задачах мы быстрее поймем, когда что-то пошло не так, и сможем быстро локализовать проблему. Про идемпотентность я уже говорил: в случае перезапуска можно быть уверенным, что мы обрабатываем ровно ту порцию данных, которую хотим обработать.
• Следующий неочевидный совет – при изменении schedule_interval или start_date нужно менять dag_id. Это связано с тем, что в базе метаданных Airflow уже есть запись о том, что такой-то пайплайн запускается тогда-то. При изменении расписания в таблицу DAGS добавляется еще одна строчка, что сводит с ума Scheduler, потому что он видит два пайплайна с разным расписанием. Для решения этой проблемы мы раньше указывали версию в названии пайплайнов, но сейчас мы перешли на подход, при котором вшиваем расписание прямо в dag_id. Таким образом пайплайн автоматически получает новое имя, и этим не нужно заниматься вручную.
• При создании пайплайна можно передать ему параметр catchup. Если его значение True, то Airflow начнет создавать запуски пайплайна для каждого интервала, начиная от start_date до текущей даты. Иногда это совсем не то, что вам нужно. При значении False Airflow создаст только один запуск за последний доступный интервал. Самое интересное, что по умолчанию в Airflow этот параметр равен True (значение по умолчанию задается в конфиг-файле).
• И последнее – стоит делать инициализацию пайплайнов максимально легкой. Шедулер раз в определенный промежуток времени приходит в директорию с пайплайнами, ищет все python файлы, в которых присутствуют слова airflow и DAG, и запускает код в этих файлах, после чего ищет все созданные объекты класса DAG. Тяжелая логика в файлах с пайплайнами может сильно повлиять на производительность. Например, если для инициализации пайплайна мы сначала ходим в базу данных, а она вдруг начинает висеть или таймаутить. Или сначала ходим в REST API, но вызов requests.get() без указания таймаута вдруг начинает бесконечно висеть.

Генерация дагов: генератор

from libs.dag_from_config.dag_generator import DagGenerator
from libs.runners.docker_runner import DockerRunner

generator = DagGenerator(config_dir='dag_configs/docker_runner', prefix='docker')
dags = generator.generate(task_runner=DockerRunner)

for dag in dags:
   globals()[dag.dag_id] = dag  # хак для динамической генерации дагов

docker_image = "docker_registry/attribution/calculation:1.1.0"

dags {
 attribution {
   owner = "RND"
   name = "attribution"
   start_date = "20190601"
   emails = [...]
   schedule_interval = "0 1 * * *"
   depends_on_past = true
   concurrency = 4

   description = """
   - Считывает события из z_log
   - Атрибуцирует их друг к другу по определённой логике
   - Проставляет параметры, необходимые для хранения данных
   - Записывает новую партицию за вчера
   """

   tags = ["critical"]

docker {
 connections {
   LMD_DWH = "dwh"
 }

 containers {
   desktop {
     image = ${docker_image}
     connections = [LMD_DWH]

     environment {
       LMD_YARN_QUEUE = "{{ var.value.YARN_QUEUE }}"
       LMD_INSTANCES = 60
       LMD_MEMORY_PER_INSTANCE = "4g"
       LMD_ZLOG_SOURCE = "z_log_db.z_log"
       LMD_ATTRIBUTION_TABLE = "{{ var.value.HIVE_DB_DERIVATIVES }}.z_log_attribution"
       LMD_ORDERS_TABLE = "rocket_dwh_bl.fct_orderitem_detail"
       LMD_PLATFORMS = "desktop"

       LMD_RUN_DATE = "{{ ds_nodash }}"
     }
   }
   mobile {...}
   iOS {...}
   Android {...}
 }
 tasks = [[desktop, mobile, iOS, Android]]
}

  awaits {
     z_log_compaction {
       dag = "compactor_daily_23_21_A_A_A"
       task = "compact_z_log_db_z_log"
       timedelta = 3hr37m
     }
     oracle_bl_fct_orderitem_detail {
       dag = "await_bl_fct_orderitem_detail_0_1_A_A_A"
     }
   }
 }
}

docker_image = "docker_registry/attribution/calculation:1.1.0"

dags {
 attribution {
   owner = "RND"
   name = "attribution"
   start_date = "20190601"
   emails = [...]
   schedule_interval = "0 1 * * *"
   depends_on_past = true
   concurrency = 4

   description = """
   - Считывает события из z_log
   - Атрибуцирует их друг к другу по определённой логике
   - Проставляет параметры, необходимые для хранения данных
   - Записывает новую партицию за вчера
   """

   tags = ["critical"]


   docker {
     connections {
       LMD_DWH = "dwh"
     }

     containers {
       desktop {
         image = ${docker_image}
         connections = [LMD_DWH]

         environment {
           LMD_YARN_QUEUE = "{{ var.value.YARN_QUEUE }}"
           LMD_INSTANCES = 60
           LMD_MEMORY_PER_INSTANCE = "4g"
           LMD_ZLOG_SOURCE = "z_log_db.z_log"
           LMD_ATTRIBUTION_TABLE = "{{ var.value.HIVE_DB_DERIVATIVES }}.z_log_attribution"
           LMD_ORDERS_TABLE = "rocket_dwh_bl.fct_orderitem_detail"
           LMD_PLATFORMS = "desktop"

           LMD_RUN_DATE = "{{ ds_nodash }}"
         }
       }
       mobile {...}
       iOS {...}
       Android {...}
     }
     tasks = [[desktop, mobile, iOS, Android]]
   }


   awaits {
     z_log_compaction {
       dag = "compactor_daily_23_21_A_A_A"
       task = "compact_z_log_db_z_log"
       timedelta = 3hr37m
     }
     oracle_bl_fct_orderitem_detail {
       dag = "await_bl_fct_orderitem_detail_0_1_A_A_A"
     }
   }
 }
}

• 2 пункта в блоке awaits превратились в два сенсора, которые дожидаются выполнения базового пайплайна,
• 4 задачи, которые мы указали в блоке docker, превратились в 4 параллельно бегущие задачи,
• между двумя блоками операторов мы добавили DummyOperator, чтобы не было паутины из связей между задачами.

Что мы хотим делать дальше

Почему мы выбрали Airflow

1. Во-первых, это Python, где с помощью двух циклов и пары условий можно сделать элегантный, правильно работающий пайплайн. И его не нужно будет описывать огромным куском XML. Плюс из коробки доступна практически вся экосистема Python и весь его зоопарк библиотек, который можно использовать как угодно.
2. Отсутствие XML сильно упрощает code review. Мы написали код пайплайна и конфиги к нему, и всё замечательно, всё работает. На самом деле, можно затащить XML или любой другой формат конфигов, но это уже дело вкуса.
3. Можно прикрутить базовые unit-тесты, которые будут проверять как отдельный пайплайн, так и конкретные операторы.
4. Есть куча кастомизаций, и когда я говорю «куча», я немножко преуменьшаю. В Airflow можно кастомизировать стандартные операторы под ваши нужды. Он также не будет против, если мы вдруг захотим реализовать свой оператор, сенсор или хук.
5. Airflow может ходить в огромное число внешних источников (и некоторые даже считают это минусом).
6. Возможна интеграция с Active Directory и гибкая настройка доступов с помощью RBAC (role-based access control, подробнее об этом в минусах)
7. Worker могут масштабироваться за счёт использования Celery или Kubernetes.
8. Поскольку это open source-штука, можно в любой момент залезть в исходник любого оператора или в исходники шедулера и посмотреть, что там на самом деле происходит.
9. У Airflow прекрасная документация, где описаны все подводные камни. Также о нём написано большое количество гайдов и есть активное сообщество.
10. Доступен мониторинг и алертинг: из коробки есть интеграция со statsd для отправки метрик, интеграция с Sentry – мы попробовали ее прикрутить, но Airflow слал нам алерты на каждый чих, и мы решили пока отложить это дело. Также есть Airflow-exporter для интеграции с Prometheus.

Минусы Airflow, которые мы обнаружили

1. Самый большой минус – это порог вхождения: необходимо настроить окружение, понять работу компонентов, как работает идемпотентность и почему execution_date в ежедневном даге – это на самом деле вчера, а не сегодня.
2. Для кого-то может быть минусом невозможность накидывания пайплайна через веб-интерфейс, как это сделано, например, в Apache NiFi. Но мы считаем это плюсом – code-review с подсветкой diff-а организовать проще, чем ревью графических схем.
3. Также Scheduler иногда причиняет некоторое количество боли из-за задержки между запусками задач.
4. У кастомизации есть обратная сторона медали – некоторые операторы могут работать не так, как мы ожидаем. Или у них не будет достаточной функциональности для наших задач – в этих случаях придется переписать или расширить стандартные операторы.
5. По умолчанию у Airflow довольно скромная ролевая модель: доступны роли пользователя и админа. При этом авторизованный пользователь имеет полный доступ к пайплайнам, включая те, которыми он не владеет. RBAC позволяет гибко настраивать роли пользователей (список доступных ролей) и их доступы, а также открывает бонусные плюшки в UI (например, указывать для пайплайнов теги, по которым можно фильтроваться). Однако подключение RBAC – отдельная веселая история с security настройками Flask, почитать об этом можно тут.
6. Для полноценной локальной разработки и тестирования нужно поднять у себя все эти компоненты: шедулер, воркер, веб-сервер, очередь задач, базу метаданных. Когда я запускаю весь этот зоопарк у себя в докере, мой ноутбук превращается во взлетающий вертолет.

В каких случаях можно подружиться с Airflow

• Когда crontab’a уже не хватает и нужен cron на батарейках.
• Если человек уже дружит с Python.
• Если уже что-то крутится в Docker, но не хватает дополнительных свистелок типа интерфейса, просмотра статистики и зависимостей между задачами.
• Если не критична задержка между запусками задач, которую я упоминал, и если не критичен real time.
• Также Airflow подойдет, если потребности описываются словами “окей, нужно раз в день запустить штуку Х, также нужно два раза в день запускать штуку У, после чего запускать штуку Z – только если выполнится это условие”.

Что можно почитать про Airflow

• В документации описан процесс настройки и часто задаваемые вопросы – ссылка
• Также есть компания Astronomer, которая занимается hosted-установкой Airflow в облаке или в Kubernetes. Также она делает прикольные гайды – ссылка
• Плюс внезапно Astronomer делает подкаст про Airflow – ссылка
• Русскоязычный чатик Airflow (ссылка) и англоязычный Slack (ссылка).