如何从Dataframe自动生成Airflow任务依赖关系？

问题根源 #

你之前的代码存在两个核心问题：

1. 生成任务时，每次循环用变量覆盖之前的任务对象，没有保留所有任务的引用；
2. 配置依赖时，对整数ID使用>>操作符，但Airflow的>>是Operator对象的专属方法，对整数调用完全无效。

解决步骤 #

1. 用字典存储所有任务对象 #

修改任务生成逻辑，把每个任务对象存入字典，键用任务ID（和Dataframe中的id_process对应），方便后续快速定位任务：

# 初始化字典，映射任务ID到Operator对象
task_map = {}

for index, row in tasks.DF_PROCESS_LIST.iterrows():
    task_id = str(row["id_process"])  # 统一转为字符串，避免类型不匹配
    # 创建任务对象
    current_task = DummyOperator(task_id=task_id, dag=dag)
    # 将任务存入字典
    task_map[task_id] = current_task

2. 遍历依赖Dataframe，建立任务依赖 #

从字典中取出对应的任务对象，使用>>操作符配置依赖。注意顺序：上游任务 >> 下游任务，即下游任务依赖上游任务完成。

for index, row in tasks.DF_DEPENDENCES.iterrows():
    # 转换ID类型，和字典的键保持一致
    downstream_task_id = str(row['process_id'])
    upstream_task_id = str(row['depends_on_id'])
    
    # 校验任务是否存在，避免KeyError
    if downstream_task_id in task_map and upstream_task_id in task_map:
        # 建立依赖关系
        task_map[upstream_task_id] >> task_map[downstream_task_id]
    else:
        # 可选：打印警告，提示无效依赖
        print(f"警告：任务 {downstream_task_id} 或 {upstream_task_id} 不存在，跳过该依赖配置")

额外优化建议 #

• 标准化表格模板：给非专业用户提供固定列名的表格（比如id_process、任务名称、depends_on_id），用户只需填写任务ID和依赖ID，无需修改代码；
• 支持多依赖：如果一个任务需要依赖多个上游任务，可在depends_on_id列用逗号分隔多个ID，拆分后处理：

  for index, row in tasks.DF_DEPENDENCES.iterrows():
      downstream_task_id = str(row['process_id'])
      upstream_ids = str(row['depends_on_id']).split(',')
      
      if downstream_task_id in task_map:
          for upstream_id in upstream_ids:
              upstream_id = upstream_id.strip()
              if upstream_id in task_map:
                  task_map[upstream_id] >> task_map[downstream_task_id]
  

• 替换打印为日志：生产环境中用Airflow日志模块替代print，方便排查问题。

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