我想在 Airflow 中创建一个条件任务,如下面的架构中所述.预期的情况如下:
I would like to create a conditional task in Airflow as described in the schema below. The expected scenario is the following:
- 任务 1 执行
- 如果任务 1 成功,则执行任务 2a
- Else 如果任务 1 失败,则执行任务 2b
- 最终执行任务 3
以上所有任务都是 SSHExecuteOperator.我猜我应该使用 ShortCircuitOperator 和/或 XCom 来管理条件,但我不清楚如何实现它.你能描述一下解决方案吗?
All tasks above are SSHExecuteOperator. I'm guessing I should be using the ShortCircuitOperator and / or XCom to manage the condition but I am not clear on how to implement that. Could you please describe the solution?
推荐答案你必须使用 气流触发规则
所有操作符都有一个 trigger_rule 参数,它定义了触发生成任务的规则.
All operators have a trigger_rule argument which defines the rule by which the generated task get triggered.
触发规则的可能性:
ALL_SUCCESS = 'all_success' ALL_FAILED = 'all_failed' ALL_DONE = 'all_done' ONE_SUCCESS = 'one_success' ONE_FAILED = 'one_failed' DUMMY = 'dummy'这是解决您问题的想法:
Here is the idea to solve your problem:
from airflow.operators.ssh_execute_operator import SSHExecuteOperator from airflow.utils.trigger_rule import TriggerRule from airflow.contrib.hooks import SSHHook sshHook = SSHHook(conn_id=<YOUR CONNECTION ID FROM THE UI>) task_1 = SSHExecuteOperator( task_id='task_1', bash_command=<YOUR COMMAND>, ssh_hook=sshHook, dag=dag) task_2 = SSHExecuteOperator( task_id='conditional_task', bash_command=<YOUR COMMAND>, ssh_hook=sshHook, dag=dag) task_2a = SSHExecuteOperator( task_id='task_2a', bash_command=<YOUR COMMAND>, trigger_rule=TriggerRule.ALL_SUCCESS, ssh_hook=sshHook, dag=dag) task_2b = SSHExecuteOperator( task_id='task_2b', bash_command=<YOUR COMMAND>, trigger_rule=TriggerRule.ALL_FAILED, ssh_hook=sshHook, dag=dag) task_3 = SSHExecuteOperator( task_id='task_3', bash_command=<YOUR COMMAND>, trigger_rule=TriggerRule.ONE_SUCCESS, ssh_hook=sshHook, dag=dag) task_2.set_upstream(task_1) task_2a.set_upstream(task_2) task_2b.set_upstream(task_2) task_3.set_upstream(task_2a) task_3.set_upstream(task_2b)更多推荐
如何在 Airflow 中创建条件任务
发布评论