Airflow 2.4.3中DataprocOperator无法获取Jinja变量project_id问题

在Airflow 2.4.3版本中使用DataprocInstantiateWorkflowTemplateOperator时，无法通过{{var.value.project_id}}获取Airflow变量project_id的值，执行任务时出现错误。使用的是Google官方提供的示例代码：

"""Example Airflow DAG that kicks off a Cloud Dataproc Template that runs a
Spark Pi Job.

This DAG relies on an Airflow variable
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id - Google Cloud Project ID to use for the Cloud Dataproc Template.
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocInstantiateWorkflowTemplateOperator,
)
from airflow.utils.dates import days_ago

project_id = "{{var.value.project_id}}"


default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "project_id": project_id,
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    # The id you will see in the DAG airflow page
    "dataproc_workflow_dag",
    default_args=default_args,
    # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:

    start_template_job = DataprocInstantiateWorkflowTemplateOperator(
        # The task id of your job
        task_id="dataproc_workflow_dag",
        # The template id of your workflow
        template_id="sparkpi",
        project_id=project_id,
        # The region for the template
        region="us-central1",
    )

执行时Airflow UI报错，核心问题是project_id参数被传入了原始字符串{{var.value.project_id}}，而非实际配置的项目ID值。

1. 模板渲染规则限制：Airflow的Jinja模板语法{{var.value.project_id}}仅在任务执行阶段对运算符声明为支持模板的字段进行渲染。而Airflow 2.x版本中，DataprocInstantiateWorkflowTemplateOperator的project_id参数不在默认的template_fields列表中，因此不会触发模板解析，直接将字符串原样传入。
2. 示例版本适配问题：Google提供的该示例大概率是针对Airflow 1.x编写的，旧版本中部分运算符参数默认支持模板渲染，而Airflow 2.x对模板字段做了更严格的区分，导致旧示例代码在新版本中失效。

有两种可行的修复方式：

方法1：直接通过Airflow变量API读取值 #

在DAG解析阶段直接获取变量的实际值，无需依赖Jinja模板：

from airflow.models import Variable

# 替换原有的project_id赋值语句
project_id = Variable.get("project_id")

方法2：强制为参数启用模板渲染 #

如果需要保留Jinja模板语法，可以显式指定project_id为需要渲染的字段：

start_template_job = DataprocInstantiateWorkflowTemplateOperator(
    task_id="dataproc_workflow_dag",
    template_id="sparkpi",
    project_id="{{var.value.project_id}}",
    region="us-central1",
    # 把project_id加入模板渲染字段列表
    template_fields=DataprocInstantiateWorkflowTemplateOperator.template_fields + ("project_id",)
)

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