Эффективный способ развертывания dag файлов в потоке воздуха

Существуют ли какие-либо рекомендации, которые следует применять для развертывания новых даг в воздушном потоке?

Я видел пару комментариев на форуме google, в котором говорится, что dags сохраняются внутри репозитория GIT, и то же самое периодически синхронизируется с локальным местоположением в кластере воздушного потока.
Что касается этого подхода, у меня было несколько вопросов

Поддерживаем ли мы отдельные файлы dag для отдельных сред? (тестирование) Как обрабатывать откат ETL для более старой версии, если в новой версии есть ошибка?

Любая помощь здесь очень ценится. Дайте мне знать, если вам нужны какие-либо подробности?

Ответ 1

Вот как мы управляем этим для нашей команды.

Во-первых, с точки зрения соглашения об именах, каждое из наших имен файлов DAG соответствует идентификатору DAG из содержимого самой DAG (включая версию DAG). Это полезно, потому что в конечном итоге это идентификатор DAG, который вы видите в пользовательском интерфейсе Airflow, чтобы вы точно знали, какой файл использовался за каждым DAG.

Пример для DAG, подобный этому:

from airflow import DAG
from datetime import datetime, timedelta

default_args = {
  'owner': 'airflow',
  'depends_on_past': False,
  'start_date': datetime(2017,12,05,23,59),
  'email': ['[email protected]'],
  'email_on_failure': True
}

dag = DAG(
  'my_nice_dag-v1.0.9', #update version whenever you change something
  default_args=default_args,
  schedule_interval="0,15,30,45 * * * *",
  dagrun_timeout=timedelta(hours=24),
  max_active_runs=1)
  [...]

Имя файла DAG будет следующим: my_nice_dag-v1.0.9.py

Все наши файлы DAG хранятся в репозитории Git (помимо прочего)
Каждый раз, когда в нашей основной ветке выполняется запрос на слияние, наш конвейер непрерывной интеграции запускает новую сборку и упаковывает наши файлы DAG в zip-архив (мы используем Atlassian Bamboo, но есть и другие решения, такие как Jenkins, Circle CI, Travis...)
В Bamboo мы настроили сценарий развертывания (оболочку), который распаковывает пакет и помещает файлы DAG на сервер Airflow в папку /dags.
Обычно мы тестируем группы DAG в DEV, затем в UAT и, наконец, в PROD. Развертывание выполняется одним нажатием кнопки в пользовательском интерфейсе Bamboo благодаря вышеописанному сценарию оболочки.

Преимущества

Поскольку вы включили версию DAG в свое имя файла, предыдущая версия файла DAG не перезаписывается в папке DAG, поэтому вы можете легко вернуться к ней
Когда ваш новый файл DAG загружается в Airflow, вы можете узнать его в пользовательском интерфейсе благодаря номеру версии.
Поскольку ваше имя файла DAG = DAG Id, вы даже можете улучшить сценарий развертывания, добавив некоторую командную строку Airflow для автоматического включения новых групп DAG после их развертывания.
Поскольку каждая версия групп доступности баз данных историзирована в Git, мы всегда можем вернуться к предыдущим версиям, если это необходимо.

Ответ 2

Привет Алексис и другие.

Мы следовали вашему совету, но столкнулись с проблемами, так как мы используем пакеты Python и когда у нас в Airflow развернуто две версии одной и той же группы доступности баз данных, пакеты конфликтуют. Мы используем упакованные DAG.

my_dag-1.0.0.zip
    my_dag.py
        my_package/__init__.py
        my_package/functions.py

При развертывании my_dag-2.0.0 в параллель, похоже, загружается my_package из my_dag-1.0.0. Как вы решаете это?

Спасибо мартин