自定义气流运算符中的意外 Jinja 模板行为答案

【问题标题】：Unexpected Jinja Template Behaviour in Custom Airflow Operator自定义气流运算符中的意外 Jinja 模板行为
【发布时间】：2022-11-13 20:20:57
【问题描述】：

我在 Airflow 中制作了一个自定义传感器，它继承了BashSensor。

传感器：

class MySensor(BashSensor):
    def __init__(self, time, **kwargs):  # {{ ts }} is passed as time in the DAG
        self.time = time
        cmd = f"java some-other-stuff {self.time}"  # rendered/correct value for self.time
        super().__init__(**kwargs, bash_command=cmd)

    def poke(self, context):
        status = super().poke()  # returns True or False
        if status:
            print(self.time)  # {{ ts }} is printed instead of rendered value
        else:
            print("trying again")
        return status

当我查看 DAG 中操作员任务的渲染选项卡时，我看到 bash_command 具有正确的渲染值（{{ ts }} 作为 time 传递）。

问题是每当调用 poke 并返回 True 时，我在 print 语句中看到 {{ ts }} 而不是呈现的值。

当我在poke 函数中打印它时，我希望self.time 具有呈现的值（一些时间戳）而不是{{ ts }}。

【问题讨论】：

标签： airflow jinja2 mwaa

【解决方案1】：

cmd 和 time 在您的代码中都不是模板字段，因此 Jinja 引擎不会处理它们。您看到命令被模板化的原因是因为在 super 调用中您执行以下操作：

bash_command=cmd

bash_command 是 BashSensor 的模板字段因此，当命令按预期解析为正确的字符串时，创建它的各个组件不包含渲染值。更详细地解释：time = "{{ ds }}" 将始终保留为该字符串。它永远不会被渲染。

当你执行 cmd = f"java some-other-stuff {self.time}" 时，它会变成：

"java some-other-stuff {{ ds }}"

此字符串分配给模板化字段 bash_command，当执行代码时呈现 {{ ds }} 的值。

要解决您的问题，您只需将要模板化的参数添加到序列中：

class MySensor(BashSensor):
    ...
    template_fields = template_fields: Sequence[str] = tuple({'time'} | set(BashSensor.template_fields))
    ...

【讨论】：