airflow trigger rule 'none_failed' doesn't work

Viewed 617

I have a workflow below. What I want to do is to run the join task if all parents in success or skipped state. For this I use none_failed trigger rule, but have a problem when all upstream tasks in skipped state, so downstream task join is skipped automatically. Is there any solution, so I can run join task if all parents are skipped? Trigger rule all_done doesn't fit for me. Airflow workflow

dag = DAG(
    dag_id="branch_without_trigger",
    schedule_interval="@once",
    start_date=dt.datetime(2019, 2, 28),
)


def fun():
    raise AirflowSkipException("exception")


run_this_first = DummyOperator(task_id="run_this_first", dag=dag)
branching = DummyOperator(task_id="branching", dag=dag, python_callable=fun)

branch_a = DummyOperator(task_id="branch_a", dag=dag)
follow_branch_a = PythonOperator(
    task_id="follow_branch_a", dag=dag, python_callable=fun
)

branch_b = PythonOperator(task_id="branch_b", dag=dag, python_callable=fun)

join = DummyOperator(task_id="join", dag=dag, trigger_rule="none_failed")

run_this_first.set_downstream(branching)
branching >> branch_a >> follow_branch_a >> join
branching >> branch_b >> join

0 Answers
Related