AirflowにはBranchPythonOperatorがあり、これを使うことでより直接的に分岐の依存関係を表現することができます。
docs](https://airflow.incubator.apache.org/concepts.html?highlight=branch#branching)にその使い方が書かれています。
BranchPythonOperator は PythonOperator と似ていますが、task_id を返す python_callable を期待する点が異なります。返されたtask_idが追従し、他のすべてのパスはスキップされます。Python関数が返すtask_idは、BranchPythonOperatorタスクの直接下流のタスクを参照する必要があります。
...
もし、いくつかのタスクをスキップしたい場合、パスが空であってはいけないことに注意し、その場合はダミーのタスクを作ります。
def dummy_test():
return 'branch_a'
A_task = DummyOperator(task_id='branch_a', dag=dag)
B_task = DummyOperator(task_id='branch_false', dag=dag)
branch_task = BranchPythonOperator(
task_id='branching',
python_callable=dummy_test,
dag=dag,
)
branch_task >> A_task
branch_task >> B_task
Airflowのバージョン >=1.10.3 をインストールしている場合、タスクIDのリストを返す ことで、1つのOperatorで複数の下流パスをスキップしたり、結合前のダミータスクを使用しない ことも可能です。
気流トリガールール]1を使用する必要があります。
すべてのオペレータは、生成されたタスクがトリガーされるルールを定義するtrigger_rule引数を持っています。
トリガールールの可能性。
ALL_SUCCESS = 'all_success'
ALL_FAILED = 'all_failed'
ALL_DONE = 'all_done'
ONE_SUCCESS = 'one_success'
ONE_FAILED = 'one_failed'
DUMMY = 'dummy'
ここで、あなたの問題を解決するためのアイデアを紹介します。
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)