Загрузить DAG-файлы в Apache Airflow™

DAG-файлы поступают в кластер из Git-репозитория или S3-хранилища, выбранного при создании кластера. В этой инструкции используется простой DAG для ручного запуска в Apache Airflow™ 3.1.8.

Перед началомПеред началом

1. Убедитесь, что кластер находится в состоянии Ready. При необходимости возобновите его.
2. В консоли управления Stackland выберите проект и пространство имен. В левом меню выберите Кластеры Airflow. Если пункт скрыт, раскройте Ещё. Откройте нужный кластер.
3. На вкладке Основное проверьте раздел Источник DAG: репозиторий и Git ref либо бакет, а также путь внутри источника.
4. Подготовьте права на запись в выбранный Git-репозиторий или бакет. Учетные данные администратора веб-интерфейса Apache Airflow™ не предоставляют эти права.

Подготовить DAG-файлПодготовить DAG-файл

Сохраните следующий код в файл hello_airflow.py:

from datetime import datetime, timezone

from airflow.sdk import Param, dag, get_current_context, task


@dag(
    dag_id="hello_airflow",
    start_date=datetime(2026, 1, 1, tzinfo=timezone.utc),
    schedule=None,
    catchup=False,
    params={"message": Param("Hello from Airflow!", type="string")},
    tags=["example"],
)
def hello_airflow():
    @task
    def greet():
        message = get_current_context()["params"]["message"]
        print(message)
        return message

    greet()


hello_airflow()

Этот DAG содержит одну задачу greet: она записывает сообщение в лог и возвращает его как результат XCom. По умолчанию сообщение — Hello from Airflow!; при ручном запуске его можно передать в параметре message. Значение schedule=None отключает запуски по расписанию. Дополнительные пакеты для примера не требуются.

Если ваши DAG используют библиотеки, которых нет в образе кластера, сначала добавьте зависимости и дождитесь готовности кластера.

Загрузить из GitЗагрузить из Git

1. Добавьте hello_airflow.py в каталог репозитория, указанный в поле Путь к DAG. Если поле пустое, используйте корень репозитория.

2. Сохраните изменение в коммите и отправьте его в ветку, указанную в Git ref.

   Если кластер настроен на конкретный тег или коммит, измените Git ref на ревизию, содержащую новый файл.

3. Дождитесь синхронизации и проверьте появление DAG.

URL репозитория и путь к DAG после создания кластера не изменяются. Для другого источника создайте новый кластер.

Загрузить из S3Загрузить из S3

Загрузите файл в бакет и под префикс, указанные в источнике DAG. Например, для пути dags/ ключ объекта должен быть dags/hello_airflow.py.

Через консоль управления StacklandЧерез консоль управления Stackland

Этот способ подходит для бакета Object Storage в Stackland.

1. Если бакет создавался автоматически, найдите его имя в карточке кластера: Основное → Связанные ресурсы → Bucket для DAG.
2. В том же проекте и пространстве имен откройте Object Storage, выберите нужный бакет и вкладку Объекты.
3. Откройте папку, соответствующую пути внутри источника. Если папки нет, раскройте меню рядом с Загрузить объект, выберите Создать папку, введите имя и нажмите Создать. Затем откройте созданную папку. При пустом пути загружайте файл в корень бакета.
4. Нажмите Загрузить объект, выберите hello_airflow.py и нажмите Загрузить.
5. Дождитесь сообщения об успешной загрузке и проверьте появление DAG.

Во внешнее S3-хранилищеВо внешнее S3-хранилище

Используйте консоль вашего S3-сервиса или настроенный S3-клиент с правами на запись в бакет. Например, для AWS CLI с профилем airflow-dags:

aws --profile airflow-dags \
  --endpoint-url=https://<эндпоинт_S3> \
  --region=<регион> \
  s3 cp hello_airflow.py s3://<имя_бакета>/dags/hello_airflow.py

Замените эндпоинт, регион, имя бакета и путь на значения своего хранилища. Если источник DAG настроен на корень бакета, не добавляйте dags/ к ключу объекта. Подробнее о команде — в справочнике AWS CLI.

Проверить появление DAGПроверить появление DAG

1. Откройте веб-интерфейс Apache Airflow™.
2. Перейдите в раздел Dags и найдите hello_airflow.
3. Откройте DAG и на вкладке Code проверьте загруженный код.

Git и S3 синхронизируются раз в 60 секунд. После этого требуется время на разбор файла, поэтому DAG появляется не сразу. Состояние кластера Ready само по себе не подтверждает загрузку DAG.

Если DAG не появился, проверьте выбранные репозиторий или бакет, путь, Git ref и права доступа. Для диагностики используйте разделы Ошибки Git, Ошибки S3-синхронизации DAG и Ошибки импорта DAG.

После появления DAG запустите его.

Обновить DAGОбновить DAG

1. Измените Python-файл. Чтобы обновить существующий DAG, сохраните его dag_id: другое значение создаст отдельный DAG.

2. Опубликуйте изменение в настроенном источнике:

   • Для Git отправьте коммит в выбранную ветку. Если используется тег или конкретный коммит, измените Git ref.
   • Для S3 загрузите новую версию файла с тем же ключом объекта, включая префикс. При запросе подтверждения разрешите замену объекта.

3. Дождитесь синхронизации и разбора файла. В Apache Airflow™ откройте Dags → нужный DAG → Code и проверьте измененный фрагмент. Если интерфейс предлагает выбор версии, выберите последнюю.

4. Выполните ручной запуск и проверьте результат и логи. Для выбранного запуска или задачи вкладка Code позволяет проверить код, связанный именно с этим выполнением.

Завершенные задачи не повторяются автоматически после обновления файла. Дальнейшие запуски определяются расписанием или ручным запуском. Если требуется повторить задачу в существующем запуске, используйте Clear.

Чтобы вернуть прежний код, восстановите файл в Git или S3 либо укажите прежний Git ref и повторите проверку. Это не отменяет изменений, которые задачи уже внесли во внешние системы.

Удалить DAGУдалить DAG

Если нужно временно прекратить запуски, достаточно приостановить DAG.

Чтобы убрать DAG из источника и сохранить историю:

1. Выберите время, когда у DAG нет активных запусков, и приостановите его. Приостановка сама по себе не завершает выполняющиеся задачи.

2. Удалите определение DAG из Python-файла или удалите весь файл, если в нем нет других нужных DAG:

   • Для Git отправьте изменение в настроенную ветку; при необходимости обновите Git ref кластера.
   • Для S3 удалите соответствующий объект из настроенного бакета и префикса.

3. Дождитесь синхронизации и обработки изменений. Apache Airflow™ пометит DAG как отсутствующий в источнике. Его история останется в базе метаданных до плановой очистки.

Для удаления истории сначала сохраните нужные результаты и логи. В разделе Dags найдите DAG, при необходимости изменив фильтры списка, откройте его меню действий и выберите Delete Dag. Прочитайте предупреждение и подтвердите удаление кнопкой Delete. Будут удалены метаданные DAG, его запусков и задач. Эта операция не удаляет Python-файл из Git или S3 и не отменяет обработку данных во внешних системах.

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