Airflow DAG定义依赖时设置参数实现任务复用的最佳方案

你可以通过两种主流方案解决重复创建任务的问题，完全符合DRY原则，后续修改配置仅需调整一处即可全局生效：

方案1：循环动态生成任务（兼容所有Airflow版本） #

将参数统一维护在列表中，通过循环批量生成任务，公共配置可一次性全局定义：

# 统一维护所有参数，新增/修改类型仅需调整该列表
THING_TYPES = ["A", "B"]
# 用字典存储生成的任务，方便后续设置依赖时调用
do_thing_tasks = {}

# 公共任务配置统一管理，修改时无需逐个调整
common_task_params = {
    "python_callable": do_thing,
    "retries": 2,
    "retry_delay": timedelta(minutes=5)
}

for thing_type in THING_TYPES:
    do_thing_tasks[thing_type] = PythonOperator(
        task_id = f"do_thing_type_{thing_type.lower()}",
        op_kwargs = {"thing_type": thing_type},
        **common_task_params
    )

# 设置依赖示例：所有do_thing任务完成后执行下游任务downstream_task
# 如果需要单独依赖某参数的任务，直接取do_thing_tasks["A"]即可
list(do_thing_tasks.values()) >> downstream_task

方案2：动态任务映射（Airflow 2.3+ 官方原生推荐） #

Airflow 2.3版本推出的动态任务映射特性专门针对该场景，无需手动写循环，直接通过原生API实现：

# 定义基础任务模板，所有公共配置统一设置
base_do_thing = PythonOperator.partial(
    task_id = "do_thing_type",
    python_callable = do_thing,
    retries = 2,
    retry_delay = timedelta(minutes=5)
)

# 传入参数列表自动生成对应数量的任务实例
do_thing_tasks = base_do_thing.expand(
    op_kwargs=[{"thing_type": "A"}, {"thing_type": "B"}]
)

# 下游任务直接依赖映射后的任务对象即可，会自动等待所有参数的实例执行完成
do_thing_tasks >> downstream_task

如果下游任务需要获取不同参数实例的输出，可直接通过do_thing_tasks.output[索引]或do_thing_tasks.map()方法处理对应实例的返回值。

内容的提问来源于stack exchange，提问作者Duck Hunt Duo