Как изменить приоритет задачи в очереди исполнителя Apache AirFlow, на что влияет метод определения весов, каким образом можно балансировать нагрузку с помощью пулов и зачем настраивать количество слотов.
Как приоритизировать задачи в очереди Apache AirFlow
Дата-инженеры, которые используют Apache AirFlow для оркестрации пакетных процессов, знают, что задачи скапливаются в очереди исполнителя. Исполнитель — это свойство конфигурации планировщика, который управляет запуском запланированных рабочих процессов. В соответствии с жизненным циклом задачи в Apache AirFlow, о чем мы писали здесь, сначала планировщик определяет время запуска задачи и других зависимостей, включая завершение вышестоящих задач. При этом задача переходит в состояние запланирована (scheduled). После назначения задачи исполнителю, она переходит в состояние очереди (queued). Когда исполнитель берет задачу и рабочий процесс начинает ее выполнение, задача переходит в соответствующее состояние выполнения (running). Иногда задачи зависают в очереди: намеренно, если достигнут установленный параллелизм, или случайно в случае сбоя планировщика.
По умолчанию все задачи, скопившиеся в очереди исполнителя, равны между собой по приоритетам и запускаются в порядке планирования, определенном в DAG. Однако, дата-инженер может вручную приоритизировать задачу с помощью параметра priority_weight, который определяет приоритеты в очереди исполнителя. По умолчанию priority_weight равен 1, но его значение можно увеличить до любого целого числа.
Каждая задача имеет истинное значение параметра priority_weight, рассчитываемое на его основе правила weight_rule и метода определения весов, используемый для расчета эффективного общего веса приоритета задачи. Всего в Apache Airflow есть 3 метода определения весов:
- downstream, когда эффективный вес задачи вычисляется как общая сумма всех нижестоящих потомков. Вышестоящие задачи будут иметь более высокий вес и будут планироваться более агрессивно при использовании положительных значений веса. Это полезно, если есть несколько экземпляров запуска DAG и нужно, чтобы все вышестоящие задачи были выполнены для всех запусков, прежде чем каждый DAG сможет продолжить обработку нижестоящих задач. Этот метод используется в Airflow по умолчанию.
- upstream, когда эффективный вес задачи равен общей сумме ее вышестоящих родительских задач. При этом последующие задачи имеют более высокий вес и будут планироваться более агрессивно при использовании положительных значений веса. Такая настройка полезна, когда есть несколько экземпляров запуска DAG и нужно, чтобы каждый из них был завершен перед запуском вышестоящих задач других запусков DAG.
- При этом методе определения весов эффективный вес задачи соответствует заданному значению priority_weight без дополнительного изменения. Это позволяет точно знать приоритет задачи и дает дополнительный эффект значительного ускорения процесса создания задач для очень больших DAG.
Начиная с версии 2.9.0 d AirFlow можно реализовать свой собственный метод определения весов, расширив класс PriorityWeightStrategy и зарегистрировав его в плагине. Эта функция пока считается экспериментальной, однако воспользоваться ей можно уже сейчас. Например, следующий код объявляет класс стратегии, которая уменьшает вес приоритета с каждой попыткой выполнения задачи DAG.
Чтобы использовать эту стратегию, следует импортировать модуль CustomPriorityWeightStrategy:
Далее можно создать экземпляр пользовательского класса и указать его в параметре задачи weight_rule:
Также можно явно указать путь к пользовательскому классу:
Управление пулами
Параметр priority_weight можно использовать вместе с пулами для ограничения параллелизма выполнения произвольных наборов задач. Пулы в Airflow используются для управления ресурсами и предотвращения перегрузки систем. Параметр pool_slots указывает, сколько слотов будет занимать каждая задача, позволяя гибко управлять нагрузкой и распределением задач.
Список пулов управляется в пользовательском интерфейсе Airflow, присваивая пулам имена и назначая им несколько рабочих слотов. Там же дата-инженер может включить отложенные задачи при подсчете занятых слотов пула. Когда задача Airflow выполняется в рамках пула, она занимает определенное количество слотов. Как только все эти слоты будут заняты, выполняемые задачи будут поставлены в очередь, и их состояние будет отображаться в пользовательском интерфейсе фреймворка. Когда слоты освобождаются, задачи в очереди начинают выполняться на основе весов приоритета задачи и ее потомков.
Если задаче не присвоен новый пул, они назначаются пулу по умолчанию default_pool, который инициализируется со 128 слотами. Пул по умолчанию может быть изменен через пользовательский интерфейс или в CLI, но не может быть удален. Например, следующий код создает 2 легковесные и 1 тяжеловесную задачи в пуле from_PG_2_Elastic:
Таким образом, пулы позволяют балансировать нагрузку, которая неодинакова у разных задач. Например, задачи связанные со сложными вычислениями или обращению к внешним системам, могут быть потреблять больше системных ресурсов. Поэтому для более эффективной эксплуатации Apache Airflow дата-инженеру полезно знать неочевидные тонкости создания задач.
Узнайте больше про администрирование и эксплуатацию Apache AirFlow для оркестрации пакетных процессов в задачах реальной дата-инженерии на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Статья: https://bigdataschool.ru/blog/news/airflow/pools-and-task-priority-in-airflow.html
Курсы: https://bigdataschool.ru/courses/data-flow-with-apache-airflow https://bigdataschool.ru/courses/airflow-on-yandex-managed-service
Наш сайт: https://bigdataschool.ru
Копирование, размножение, распространение, перепечатка (целиком или частично), или иное использование материала допускается только с письменного разрешения правообладателя ООО "УЦ Коммерсант"