Airflow从2.2.5升级至2.6.3时DAG生成报错：Cannot template BaseOperator fields: task_id

问题原因 #

• Airflow 2.3及以上版本新增了对BaseOperator核心内置字段的模板化限制，task_id作为Operator的唯一标识，属于BaseOperator的内置字段，不允许被添加到自定义Operator的template_fields中。
• 旧版本（如2.2.5）Airflow未严格校验该规则，因此之前的YamlOperator代码可正常运行；升级到2.6.3后，Airflow会直接抛出异常阻止这种违规操作。
• 从提供的YamlOperator代码可见，template_fields列表中包含了task_id，这正是触发报错的直接原因。

修复方案 #

1. 移除template_fields中的task_id #

修改YamlOperator的类定义，将task_id从template_fields中删除：

class YamlOperator(BaseOperator):
    
    template_fields = ["args", "arguments"]  # 移除"task_id"项
    trigger = None

    def __init__(self,
                 task_id: str,
                 trigger: BaseTrigger,
                 files=None,
                 arguments=None,
                 *args, **kwargs
                 ):
        super(YamlOperator, self).__init__(task_id=task_id, *args, **kwargs)
        self.task_id = task_id
        self.trigger = trigger
        self.files = files
        self.args = self.trigger.command_line_args()
        self.arguments = arguments

2. 动态task_id的替代实现（若有需求） #

如果业务场景需要动态生成task_id，不能依赖模板化机制（因为task_id必须在DAG解析阶段确定），应在实例化Operator时直接传入动态计算的值，示例：

# 基于ds动态生成task_id
my_task = YamlOperator(
    task_id=f"task_id_{{{{ ds_nodash }}}}",  # 直接在实例化时拼接动态值
    trigger=TriggerETLConfigUrl(
        "{{ ds }}",
        GCSLoader(file_path="file_path").url(),
        "{{ ds }}",
    ),
    dag=dag
)

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