У меня есть DAG, которая запускается по расписанию, и задачи внутри нее находят файл в пути, обрабатывают его и перемещают этот файл в папку архива.
Вместо того, чтобы ждать расписания, я вручную запустил DAG. Запущенная вручную группа обеспечения доступности баз данных выполнила свою первую задачу и «нашла новый файл для обработки», но перед запуском второй задачи для загрузки этого файла расписание DAG автоматически выбрало и начало обработку того же файла.
Когда начался запланированный переход, он приостановил выполнение вызванного вручную DAG.
После того как запланированная группа доступности базы данных завершила работу, она вернулась к выполнению задач из группы DAG, запускаемой вручную, что вызвало сбойное состояние, поскольку группа обеспечения доступности баз данных перемещает файлы в архив из исходного каталога, и группа планирования, назначенная вручную, начала обработку файла, который, по его мнению, был Есть «благодаря успеху и информации из первого задания».
Итак:
DAG запускается вручную
DAG Триггерная задача 1 выполнена вручную
DAG запланировано вызвано
Запланированная задача DAG 1 выполнена
Запланированная задача DAG 2 выполнена
Запланированное задание 3 DAG выполнено
DAG запланировано завершено как успех
Ошибка выполнения DAG вручную запускаемой задачи 2 (из-за перемещения файла задачи 2 по расписанию, обнаруженного в задаче 1)
При запуске DAG вручную пропускаются другие задачи из-за сбоя задачи 2.
DAG запускается полностью как Failed
Итак, мой вопрос:
Как настроить Airflow таким образом, чтобы вызовы одной и той же группы DAG выполнялись FIFO независимо от того, была ли эта группа вызвана по расписанию, вручную или триггером?