[Документация Yandex Cloud](../../../index.md) > [Yandex Cloud Stackland](../../index.md) > [Пошаговые инструкции](../index.md) > Оркестрация данных > Managed Service for Apache Airflow™ > Загрузить DAG-файлы в Apache Airflow™

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

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

## Перед началом {#before-you-begin}

1. Убедитесь, что кластер находится в состоянии `Ready`. При необходимости [возобновите его](pause-resume.md#resume).
1. В консоли управления Stackland выберите проект и пространство имен. В левом меню выберите **Кластеры Airflow**. Если пункт скрыт, раскройте **Ещё**. Откройте нужный кластер.
1. На вкладке **Основное** проверьте раздел **Источник DAG**: репозиторий и Git ref либо бакет, а также путь внутри источника.
1. Подготовьте права на запись в выбранный Git-репозиторий или бакет. Учетные данные администратора веб-интерфейса Apache Airflow™ не предоставляют эти права.

## Подготовить DAG-файл {#prepare-dag}

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

```python
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!`; при ручном запуске его можно [передать в параметре](run-dag.md#parameters) `message`. Значение `schedule=None` отключает запуски по расписанию. Дополнительные пакеты для примера не требуются.

Если ваши DAG используют библиотеки, которых нет в образе кластера, сначала [добавьте зависимости](edit-cluster.md#console-dependencies) и дождитесь готовности кластера.

## Загрузить из Git {#git}

1. Добавьте `hello_airflow.py` в каталог репозитория, указанный в поле **Путь к DAG**. Если поле пустое, используйте корень репозитория.
1. Сохраните изменение в коммите и отправьте его в ветку, указанную в **Git ref**.

    Если кластер настроен на конкретный тег или коммит, [измените Git ref](edit-cluster.md#console) на ревизию, содержащую новый файл.

1. Дождитесь синхронизации и [проверьте появление DAG](#check-dag).

URL репозитория и путь к DAG после создания кластера не изменяются. Для другого источника [создайте новый кластер](create-cluster.md).

## Загрузить из S3 {#s3}

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

### Через консоль управления Stackland {#console-s3}

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

1. Если бакет создавался автоматически, найдите его имя в карточке кластера: **Основное** → **Связанные ресурсы** → **Bucket для DAG**.
1. В том же проекте и пространстве имен откройте **Object Storage**, выберите нужный бакет и вкладку **Объекты**.
1. Откройте папку, соответствующую пути внутри источника. Если папки нет, раскройте меню рядом с **Загрузить объект**, выберите **Создать папку**, введите имя и нажмите **Создать**. Затем откройте созданную папку. При пустом пути загружайте файл в корень бакета.
1. Нажмите **Загрузить объект**, выберите `hello_airflow.py` и нажмите **Загрузить**.
1. Дождитесь сообщения об успешной загрузке и [проверьте появление DAG](#check-dag).

### Во внешнее S3-хранилище {#external-s3}

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

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

Замените эндпоинт, регион, имя бакета и путь на значения своего хранилища. Если источник DAG настроен на корень бакета, не добавляйте `dags/` к ключу объекта. Подробнее о команде — в [справочнике AWS CLI](https://docs.aws.amazon.com/cli/latest/reference/s3/cp.html).

## Проверить появление DAG {#check-dag}

1. [Откройте веб-интерфейс Apache Airflow™](connect-ui.md).
1. Перейдите в раздел **Dags** и найдите `hello_airflow`.
1. Откройте DAG и на вкладке **Code** проверьте загруженный код.

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

Если DAG не появился, проверьте выбранные репозиторий или бакет, путь, Git ref и права доступа. Для диагностики используйте разделы [Ошибки Git](troubleshooting.md#git), [Ошибки S3-синхронизации DAG](troubleshooting.md#s3-dags) и [Ошибки импорта DAG](troubleshooting.md#dag-import).

После появления DAG [запустите его](run-dag.md).

## Обновить DAG {#update-dag}

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

    * Для Git отправьте коммит в выбранную ветку. Если используется тег или конкретный коммит, [измените Git ref](edit-cluster.md#console).
    * Для S3 загрузите новую версию файла с тем же ключом объекта, включая префикс. При запросе подтверждения разрешите замену объекта.

1. Дождитесь синхронизации и разбора файла. В Apache Airflow™ откройте **Dags** → нужный DAG → **Code** и проверьте измененный фрагмент. Если интерфейс предлагает выбор версии, выберите последнюю.
1. Выполните [ручной запуск](run-dag.md#trigger) и проверьте [результат и логи](view-logs.md). Для выбранного запуска или задачи вкладка **Code** позволяет проверить код, связанный именно с этим выполнением.

Завершенные задачи не повторяются автоматически после обновления файла. Дальнейшие запуски определяются расписанием или ручным запуском. Если требуется повторить задачу в существующем запуске, используйте [Clear](run-dag.md#rerun-task).

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

## Удалить DAG {#delete-dag}

Если нужно временно прекратить запуски, достаточно [приостановить DAG](run-dag.md#pause-dag).

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

1. Выберите время, когда у DAG нет активных запусков, и приостановите его. Приостановка сама по себе не завершает выполняющиеся задачи.
1. Удалите определение DAG из Python-файла или удалите весь файл, если в нем нет других нужных DAG:

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

1. Дождитесь синхронизации и обработки изменений. Apache Airflow™ пометит DAG как отсутствующий в источнике. Его история останется в базе метаданных до [плановой очистки](../../concepts/components/airflow.md#metadata-db).

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

Если удалить только метаданные, оставив определение DAG в источнике, DAG появится снова после разбора файла. Подробнее — в [описании жизненного цикла DAG](https://airflow.apache.org/docs/apache-airflow/3.1.8/core-concepts/dags.html#dag-pausing-deactivation-and-deletion).