Airflowで条件付きタスクを作成する方法

Airflowで下記のスキーマのような条件付きタスクを作りたいと思います。想定されるシナリオは以下の通りです。

  • タスク1が実行される
  • タスク1が成功したら、タスク2aを実行する。
  • タスク1が失敗した場合、タスク2bを実行する。
  • 最後にタスク3が実行される

条件付きタスク1です。

上記のタスクは全てSSHExecuteOperatorです。 ShortCircuitOperatorやXComを使って条件を管理すればよいのでしょうが、どのように実装すればよいのかがわかりません。解決方法を教えてください。

Alexis.Rolland
編集された質問 12日 1月 2018 в 5:36
プログラミング
python conditional-statements airflow
28日 4月 2017 в 10:49
2 ビュー

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

EDIT:

Airflowのバージョン >=1.10.3 をインストールしている場合、タスクIDのリストを返す ことで、1つのOperatorで複数の下流パスをスキップしたり、結合前のダミータスクを使用しない ことも可能です。

villasv
編集した答え 8日 3月 2019 в 2:11
解説 (2)
ソリューション

気流トリガールール]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=)

task_1 = SSHExecuteOperator(
        task_id='task_1',
        bash_command=,
        ssh_hook=sshHook,
        dag=dag)

task_2 = SSHExecuteOperator(
        task_id='conditional_task',
        bash_command=,
        ssh_hook=sshHook,
        dag=dag)

task_2a = SSHExecuteOperator(
        task_id='task_2a',
        bash_command=,
        trigger_rule=TriggerRule.ALL_SUCCESS,
        ssh_hook=sshHook,
        dag=dag)

task_2b = SSHExecuteOperator(
        task_id='task_2b',
        bash_command=,
        trigger_rule=TriggerRule.ALL_FAILED,
        ssh_hook=sshHook,
        dag=dag)

task_3 = SSHExecuteOperator(
        task_id='task_3',
        bash_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)
解説 (5)