python - 使用 TaskFlowAPI 在 Apache Airflow 中进行分支

Question

我找不到 Airflow 的 TaskFlowAPI 中的分支文档。我尝试以“Pythonic”方式执行此操作，但运行时，无论前一个任务返回的真值如何，DAG 都看不到 task_2_execute_if_true。

@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=['test'],
)
def my_dag():
    @task()
    def task_1_returns_boolean():
        # evaluate and return boolean value
        return boolean_value
    
    @task()
    def task_2_execute_if_true():
        # do_something...

    outcome_1 = task_1_returns_boolean()
    if outcome_1:
        outcome_2 = task_2_execute_if_true() 


executed = my_dag()

TaskFlowAPI 中正确的分支方式是什么？我应该再添加一个专门用于分支的函数吗？

Answer 1

源代码中有一个 DAG 示例:https://github.com/apache/airflow/blob/f1a9a9e3727443ffba496de9b9650322fdc98c5f/airflow/example_dags/example_branch_operator_decorator.py#L43 .

语法是:

from airflow.decorators import task

@task.branch(task_id="branching_task_id")
def random_choice():
    return "task_id_to_run"

它是在 Airflow 2.3.0 中引入的。