Airflow条件触发任务报错：分支任务ID无效求助

代码示例 #

to_be_triggered = EmptyOperator(task_id="to_be_triggered")

@task.branch()
def trigger_dag(**kwargs):
  config = kwargs.get("dag_run_config")

  if config.get("run_trigger") is True:
    return ["to_be_triggered"]
  return None


with DAG("example") as dag:
    dag_run_config = {
      "run_trigger": True
    }

    t0 = trigger_dag(dag_run_config=dag_run_config)
    t1 = EmptyOperator(task_id="end", trigger_rule=TriggerRule.ONE_SUCCESS)
    
    t0 >> t1

报错信息 #

Following branch {'to_be_triggered'}
Task failed with exception
AirflowException: 'branch_task_ids' must contain only valid task_ids. Invalid tasks found: {'to_be_triggered'}

用户疑问：未使用任务组，为何会出现task_id无效的情况？

• 核心原因：to_be_triggered任务定义在DAG上下文管理器（with DAG(...) as dag:）之外，没有被注册到当前DAG的任务集合中，Airflow无法识别该task_id。

• 修复步骤：

  1. 将to_be_triggered任务的定义移到DAG上下文内部
  2. 补全分支任务与目标任务、结束任务的依赖关系

修复后的代码：

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.decorators import task
from airflow.utils.trigger_rule import TriggerRule

@task.branch()
def trigger_dag(**kwargs):
  config = kwargs.get("dag_run_config")

  if config.get("run_trigger") is True:
    return ["to_be_triggered"]
  return None


with DAG("example") as dag:
    dag_run_config = {
      "run_trigger": True
    }

    to_be_triggered = EmptyOperator(task_id="to_be_triggered")
    t0 = trigger_dag(dag_run_config=dag_run_config)
    t1 = EmptyOperator(task_id="end", trigger_rule=TriggerRule.ONE_SUCCESS)
    
    t0 >> [to_be_triggered, t1]
    to_be_triggered >> t1

• 补充说明：
  • 分支任务返回的task_id必须是当前DAG上下文中已注册的任务，否则会被判定为无效ID
  • 依赖关系需完整：分支任务要同时连接to_be_triggered和end，to_be_triggered执行完成后也要指向end，配合ONE_SUCCESS触发规则，确保无论分支是否触发，end任务都能正常执行

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