如何在Apache Airflow中正确向自定义算子传递字典？

问题场景 #

我实现了一个自定义Airflow算子，用于接收参数并触发Glue作业，代码示例如下：

task = MyCustomOperator(
    task_id="my_op",
    custom_property=False,
    r_params = {
        "run_date": "{{ ds }}"
    }
)

算子内部会通过json.dumps()将r_params转换为字符串，再传递给仅接受字符串类型参数的Glue算子。但部分开启了render_template_as_native_obj=True的DAG执行时抛出错误，提示预期类型为str但实际得到dict。

推测原因是：即便提前用json.dumps()把字典转成字符串，在Airflow渲染{{ ds }}模板变量后，结果会被render_template_as_native_obj=True自动转换回字典类型，导致与Glue算子的参数要求冲突。

解决方案 #

方案1：在算子执行阶段强制转换为字符串 #

不管Airflow模板渲染后返回的是什么类型，在传递给Glue算子前再次执行json.dumps()，确保最终传递的是字符串：

class MyCustomOperator(BaseOperator):
    def __init__(self, r_params=None, **kwargs):
        super().__init__(**kwargs)
        self.r_params = r_params

    def execute(self, context):
        # 强制将参数转为JSON字符串，覆盖渲染后的类型
        processed_params = json.dumps(self.r_params)
        # 调用Glue算子并传递字符串参数
        glue_job = GlueJobOperator(
            task_id="trigger_glue",
            job_name="target_glue_job",
            arguments={"--r_params": processed_params},
            # 其他必要参数
        )
        glue_job.execute(context)

方案2：以字符串形式传递字典模板 #

直接将包含模板变量的字典写成JSON字符串形式传入算子，这样Airflow渲染后只会替换变量内容，最终结果仍为字符串，无需额外转换：

task = MyCustomOperator(
    task_id="my_op",
    custom_property=False,
    # 直接传JSON字符串模板
    r_params='{"run_date": "{{ ds }}"}'
)

此时算子内部可以直接将self.r_params传递给Glue算子，不需要再调用json.dumps()。注意确保字符串格式符合JSON规范，避免转义错误。

方案3：拆分参数，仅渲染需要的字段 #

如果字典中只有部分字段需要模板渲染，可以将这些字段单独提取为算子参数，在算子内部再组装成字典并转换为字符串。同时通过template_fields指定仅渲染这些单独字段：

class MyCustomOperator(BaseOperator):
    # 仅标记需要渲染的字段
    template_fields = ("run_date",)

    def __init__(self, run_date=None, **kwargs):
        super().__init__(**kwargs)
        self.run_date = run_date

    def execute(self, context):
        # 组装字典并转为字符串
        r_params = json.dumps({"run_date": self.run_date})
        glue_job = GlueJobOperator(
            task_id="trigger_glue",
            job_name="target_glue_job",
            arguments={"--r_params": r_params},
            # 其他必要参数
        )
        glue_job.execute(context)

调用时直接传入模板变量：

task = MyCustomOperator(
    task_id="my_op",
    custom_property=False,
    run_date="{{ ds }}"
)

这种方式完全避免了字典作为模板对象被Airflow转换为native object的问题。

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