Airflow中EMRServerlessCreateApplicationOperator的Jinja模板渲染失效问题

AI 小北

2026-6-18

问题原因

EMRServerlessCreateApplicationOperator的config参数默认不在Operator的模板化字段列表（template_fields）中，Airflow只会对Operator类显式声明为可模板化的字段执行Jinja渲染。而PythonOperator的op_kwargs是默认支持模板渲染的，所以能正常解析{{dag_run.conf['application_name']}}。

解决方案

方案1：自定义Operator扩展原类

继承EMRServerlessCreateApplicationOperator，将config添加到template_fields中：

from airflow.providers.amazon.aws.operators.emr_serverless import EMRServerlessCreateApplicationOperator

class TemplateableEMRServerlessCreateApplicationOperator(EMRServerlessCreateApplicationOperator):
    template_fields = (*EMRServerlessCreateApplicationOperator.template_fields, "config")

# 使用自定义Operator替换原Operator
TemplateableEMRServerlessCreateApplicationOperator(
    task_id="create-emrs-app",
    job_type="SPARK",
    release_label="emr-7.1.0",
    config={
         "name": "{{dag_run.conf['application_name']}}",
         # 其他配置项
    },
    dag=main_dag
)

方案2：用PythonOperator结合EMRServerlessHook实现

利用PythonOperator的模板能力先获取配置值，再通过Hook手动创建应用：

from airflow.providers.amazon.aws.hooks.emr_serverless import EMRServerlessHook
from airflow.operators.python import PythonOperator

def create_emrs_app(**context):
    application_name = context["dag_run"].conf["application_name"]
    hook = EMRServerlessHook(aws_conn_id="aws_default")
    hook.create_application(
        job_type="SPARK",
        release_label="emr-7.1.0",
        config={
            "name": application_name,
            # 其他配置项
        }
    )

PythonOperator(
    task_id="create-emrs-app",
    python_callable=create_emrs_app,
    provide_context=True,
    dag=main_dag
)