Airflow报错Task Not Found：BranchPythonOperator执行失败排查

你遇到的TaskNotFound错误，核心原因是用错了BranchPythonOperator的返回值规则：

• 这个算子的作用是返回后续要执行的任务ID（字符串或字符串列表），用来控制DAG的分支执行逻辑。
• 但你的代码里返回的是业务数据列表article_list（比如日志里的['63456df263', '6346d56', '6346']），Airflow会把这些值当成任务ID去查找对应的Task，而你的DAG里根本没有这些ID的任务，自然报错。

根据你的实际需求，分两种场景处理：

场景1：仅需获取业务数据，不需要分支逻辑 #

如果你的目的只是执行Python函数拿到数据，不需要分支执行，直接把BranchPythonOperator换成PythonOperator，用XCom传递数据即可：

修改后的代码：

from airflow.operators.python import PythonOperator  # 确保导入

with DAG(
    ...
    catchup=False
) as dag:

    def get_delta_lean_articles(**kwargs):
        logging.info(f"Total : {len(article_list)}")
        # 返回的数据会自动存入XCom
        return article_list

    start = snowsql_operator.StepOperator(
        task_id='start'
    )

    # 替换为PythonOperator
    get_lean_ids = PythonOperator(
        task_id='lean_article_ids',
        python_callable=get_delta_lean_articles,
        provide_context=True  # 可选，若需要访问上下文参数则开启
    )

    end = snowsql_operator.StepOperator(
        task_id='end'
    )

    start >> get_lean_ids >> end

后续任务如果需要使用article_list，可以通过ti.xcom_pull(task_ids='lean_article_ids')获取（ti是任务实例对象，可通过kwargs['ti']获取）。

场景2：确实需要分支执行逻辑 #

如果你的业务需要根据article_list的内容选择不同任务执行，那必须确保返回的是DAG中已定义的任务ID，比如：

with DAG(
    ...
    catchup=False
) as dag:

    def get_delta_lean_articles(**kwargs):
        article_list = [...]
        logging.info(f"Total : {len(article_list)}")
        # 根据业务逻辑返回对应任务ID
        if len(article_list) > 0:
            return 'process_articles'
        else:
            return 'skip_processing'

    start = snowsql_operator.StepOperator(task_id='start')

    get_lean_ids = BranchPythonOperator(
        task_id='lean_article_ids',
        python_callable=get_delta_lean_articles
    )

    # 定义分支对应的任务
    process_articles = snowsql_operator.StepOperator(task_id='process_articles')
    skip_processing = snowsql_operator.StepOperator(task_id='skip_processing')
    end = snowsql_operator.StepOperator(task_id='end')

    # 设置依赖关系
    start >> get_lean_ids
    get_lean_ids >> [process_articles, skip_processing] >> end

• BranchPythonOperator是用来控制任务分支的，不是用来返回业务数据的，业务数据传递请用XCom。
• 分支返回的任务ID必须是DAG中已经定义好的，否则必然触发TaskNotFound异常。

内容的提问来源于stack exchange，提问作者x89