Как запустить выбранную параллельную задачу за раз и пропустить несколько задач

У меня есть DAG с 5 параллельными задачами. Я не хочу выполнять их все за один раз. Я хочу передать некоторый флаг/значение задаче, предположим, что если для флага установлено значение True, оно запускается, а если для флага установлено значение False, оно пропускается. Можем ли мы сделать это в Airflow2. Аргументы по умолчанию очень простые, и я не запланировал DAG.

Мой DAG Flow выглядит примерно так

starttask>>5paralleltask>>endtask

И эти 5 параллельных задач я создал с помощью цикла for

Огромное спасибо заранее

Стоит ли изучать PHP в 2023-2024 годах?
Стоит ли изучать PHP в 2023-2024 годах?
Привет всем, сегодня я хочу высказать свои соображения по поводу вопроса, который я уже много раз получал в своем сообществе: "Стоит ли изучать PHP в...
Поведение ключевого слова "this" в стрелочной функции в сравнении с нормальной функцией
Поведение ключевого слова "this" в стрелочной функции в сравнении с нормальной функцией
В JavaScript одним из самых запутанных понятий является поведение ключевого слова "this" в стрелочной и обычной функциях.
Приемы CSS-макетирования - floats и Flexbox
Приемы CSS-макетирования - floats и Flexbox
Здравствуйте, друзья-студенты! Готовы совершенствовать свои навыки веб-дизайна? Сегодня в нашем путешествии мы рассмотрим приемы CSS-верстки - в...
Тестирование функциональных ngrx-эффектов в Angular 16 с помощью Jest
В системе управления состояниями ngrx, совместимой с Angular 16, появились функциональные эффекты. Это здорово и делает код определенно легче для...
Концепция локализации и ее применение в приложениях React ⚡️
Концепция локализации и ее применение в приложениях React ⚡️
Локализация - это процесс адаптации приложения к различным языкам и культурным требованиям. Это позволяет пользователям получить опыт, соответствующий...
Пользовательский скаляр GraphQL
Пользовательский скаляр GraphQL
Листовые узлы системы типов GraphQL называются скалярами. Достигнув скалярного типа, невозможно спуститься дальше по иерархии типов. Скалярный тип...
0
0
25
2
Перейти к ответу Данный вопрос помечен как решенный

Ответы 2

Я не уверен, что понимаю требование. Если вы хотите выполнить все задачи, но ограничить параллелизм, используйте max_active_tasks: ​​количество экземпляров задачи, разрешенных для одновременного запуска.

Итак, в вашем случае вам нужно установить:

from airflow import DAG
with DAG(
    dag_id='somedag',
    ...,
    max_active_tasks=1,
) as dag:
    ...

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

Спасибо, Элад, за вашу помощь. Я больше интересовался запуском выбранной задачи за раз и пропуском нескольких задач одновременно. Прошу прощения, если вопрос не очень ясен. Я понял, как это сделать. Добавление ответа, чтобы другие могли воспользоваться

Jaishree Rout 30.03.2022 15:52
Ответ принят как подходящий

5 параллельных задач, которые у меня есть, - это операторы bash. Чтобы пропустить несколько задач и запустить несколько задач, я заставил выбранную задачу завершиться неудачей и показать, что она пропущена, я использовал exit 99 с моей командой bash, например:

'echo "running";exit 99'

Есть два шага:

  • Во-первых, чтобы заставить задачу провалиться.
  • Во-вторых, пропустить задачу, если она терпит неудачу

Чтобы принудительно выполнить задачу, я добавил список из 5 флагов (True/False). если флаг имеет значение True, задача выполняется, если значение false, она принудительно завершается неудачно, а затем пропускается. за пропуск задачи я использовал exit 99.

И это работает, как я ожидал в моем случае.

Другие вопросы по теме