当 DAG 启用 render_template_as_native_obj=True 时，Jinja 模板（如 {{ ds }}）会渲染为原生 Python 对象而非字符串，导致 dict 类型参数在自定义算子中被意外解析，引发类型错误；本文介绍通过 context['params'] 在 execute() 中动态获取并序列化参数的可靠方案。
当 dag 启用 `render_template_as_native_obj=true` 时，jinja 模板（如 `{{ ds }}`）会渲染为原生 python 对象而非字符串，导致 `dict` 类型参数在自定义算子中被意外解析，引发类型错误；本文介绍通过 `context['params']` 在 `execute()` 中动态获取并序列化参数的可靠方案。
在 Apache Airflow 中，向自定义算子传递含 Jinja 模板（如 "run_date": "{{ ds }}"）的字典参数时，若 DAG 配置了 render_template_as_native_obj=True，Airflow 会在任务实例初始化阶段将整个参数字典（包括模板字段）提前解析为原生 Python 对象——此时 r_params 被解析为 dict，但其中的 {{ ds }} 已被替换为字符串值（如 "2024-02-27"），表面看似正常；然而问题在于：该解析发生在算子 __init__ 阶段，而此时 ds 等上下文变量尚未完全就绪，且无法支持运行时动态重渲染。更关键的是，当后续调用 json.dumps(r_params) 时，若 r_params 已是 dict，则序列化无误；但若因模板未被正确识别或版本差异导致部分字段仍为 Template 对象，则会抛出 TypeError: Object of type Template is not JSON serializable。
✅ 正确解法是：避免在 __init__ 中直接接收和处理带模板的字典参数，转而使用 Airflow 标准机制——params + Context。
Airflow 将 params 显式注入到每个任务的执行上下文（Context）中，且 params 始终以原始字典形式传入，不受 render_template_as_native_obj 影响；更重要的是，params 中的 Jinja 表达式会在 execute() 执行时，由 Airflow 的模板引擎在完整上下文环境中按需、安全地渲染为字符串。
以下是一个生产就绪的实现示例：
# my_custom_operator.py
import json
import logging
from airflow.models import BaseOperator
from airflow.utils.context import Context
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
class MyCustomOperator(BaseOperator):
template_fields = ("params",) # 显式声明 params 可被模板化（可选，增强兼容性）
def __init__(
self,
*,
custom_property: bool = False,
params: dict = None,
**kwargs,
):
super().__init__(**kwargs)
self.custom_property = custom_property
# 不在此处处理 params！仅保存引用
self.params = params or {}
def execute(self, context: Context) -> None:
# ✅ 安全获取并渲染 params：context['params'] 自动完成模板渲染
rendered_params = context.get("params", {})
logging.info("Rendered params: %s", rendered_params)
# ✅ 序列化为 JSON 字符串（Glue Job 所需格式）
try:
json_string = json.dumps(rendered_params)
except TypeError as e:
raise ValueError(f"Failed to serialize params to JSON: {e}")
# 示例：传递给 GlueJobOperator（或其他下游操作）
glue_task = GlueJobOperator(
task_id=f"{self.task_id}_glue",
job_name="my-glue-job",
script_location="s3://my-bucket/scripts/job.py",
aws_conn_id="aws_default",
region_name="us-east-1",
iam_role_name="GlueServiceRole",
# 注意：Glue 接受字符串形式的 Job Arguments
job_arguments={"--r_params": json_string},
)
glue_task.execute(context)在 DAG 文件中使用：
# dag_example.py
from airflow import DAG
from datetime import datetime
from my_custom_operator import MyCustomOperator
dag = DAG(
"glue_dag",
start_date=datetime(2024, 1, 1),
schedule=None,
render_template_as_native_obj=True, # 此配置不再影响 params 渲染
)
task = MyCustomOperator(
task_id="trigger_glue_job",
custom_property=False,
params={
"run_date": "{{ ds }}",
"env": "{{ var.value.environment }}",
"batch_id": "{{ run_id }}"
},
dag=dag,
)? 关键注意事项：
- 不要在 __init__ 中对 params 调用 json.dumps：此时模板未渲染，{{ ds }} 仍是字符串字面量，序列化后得到 "{{ ds }}" 而非实际日期。
- 务必在 execute() 中通过 context['params'] 获取：这是 Airflow 保证模板已安全渲染的唯一可靠时机。
- 显式声明 template_fields = ("params",) 是良好实践：它告知 Airflow 该字段需参与模板渲染流程，提升可维护性与兼容性（尤其在旧版 Airflow 中）。
- 若需支持嵌套模板或复杂结构，可结合 context.render_template() 手动渲染特定字段，但 params 本身已默认支持全量渲染。
总结：params + Context 是 Airflow 官方推荐、经生产验证的参数传递范式。它解耦了参数定义与执行时序，彻底规避 render_template_as_native_obj=True 带来的类型冲突，同时保持代码清晰、可测试、可扩展。
