airflow 的 `@task` 装饰器不支持运行时参数化，但可通过 `override()` 方法在任务实例化阶段动态设置 `pool`、`queue` 等操作符级属性，实现灵活的资源调度控制。
在 Airflow 的 TaskFlow API 中，@task 装饰器本质是将函数封装为可复用的 PythonOperator 实例模板，其底层对应一个 BaseOperator 子类。虽然装饰器语法本身是静态的（如 @task(pool="my_pool")），但 Airflow 提供了强大的 .override() 方法——它允许你在实际触发任务前，对任务实例的任意 BaseOperator 属性进行动态覆盖。
✅ 正确用法：使用 override() 动态指定 pool
from airflow.decorators import task
from airflow import DAG
from datetime import datetime
@task
def extractor_task(**kwargs):
print(f"Running extractor with pool: {kwargs.get('pool', 'default')}")
return "data"
# 在 DAG 定义中，根据业务逻辑动态计算 pool 值
with DAG(
"dynamic_pool_dag",
start_date=datetime(2024, 1, 1),
schedule=None,
catchup=False,
) as dag:
# 示例：按环境或数据源类型选择 pool
env = "prod" # 可来自 Variable, kwargs, or external config
pool_val = "high_priority_pool" if env == "prod" else "default_pool"
# 关键：调用 override() 并传入动态 pool，再立即调用 ()
extract = extractor_task.override(pool=pool_val)()
# 也可链式传递参数
# extract = extractor_task.override(pool=pool_val)(param1="value1", param2=42)⚠️ 注意：override() 返回的是一个新的任务实例构造器，必须加 () 才真正生成可调度的任务节点；仅写 extractor_task.override(pool=...) 不会创建任务。
? 底层原理简析
- @task 装饰后的函数（如 extractor_task）是一个 TaskDecorator 对象，具备 override() 接口；
- .override(...) 会返回一个 PartialTask 实例，它延迟绑定所有 operator 属性；
- 最终调用 (...) 时，Airflow 内部才基于当前 override 配置 + 函数默认配置，实例化完整的 PythonOperator。
? 支持动态覆盖的常用属性
| 属性 | 说明 | 示例 |
|---|---|---|
| pool | 指定任务所属资源池（用于并发控制） | override(pool="etl_pool") |
| queue | 指定 Celery/Kubernetes 队列 | override(queue="gpu_queue") |
| priority_weight | 影响调度优先级 | override(priority_weight=10) |
| retries / retry_delay | 覆盖重试策略 | override(retries=3, retry_delay=timedelta(seconds=30)) |
| execution_timeout | 设置执行超时 | override(execution_timeout=timedelta(hours=2)) |
? 进阶技巧：结合上下文动态赋值
你还可以在 override() 中使用 {{ macros }} 模板（需确保在支持 Jinja 的上下文中），或通过 kwargs 从上游任务传递值：
@task
def get_pool_strategy(**context):
# 根据 DAG 运行时间、conf 或变量决定 pool
execution_date = context["logical_date"]
return "nightly_pool" if execution_date.hour == 2 else "default_pool"
# 在 DAG 中组合使用
pool_choice = get_pool_strategy()
extract = extractor_task.override(pool=pool_choice)()✅ 总结：不要尝试“修改已装饰函数的属性”，而应利用 override() 在任务构建阶段注入动态配置——这是 Airflow 官方推荐、稳定且可测试的标准模式。
