当 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 的模板引擎在完整上下文环境中按需、安全地渲染为字符串。
以下是一个生产就绪的实现示例：
Apache Superset 是一个广泛采用的开源 BI 平台，用于 SQL 探索、图表构建和仪表板交付。当代理需要查询仓库数据、组装仪表板或使用成熟的分析界面解释指标而不是临时笔记本代码时，此技能非常有用。
# 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 带来的类型冲突，同时保持代码清晰、可测试、可扩展。
