python - 模板在 Airflow 的 on_failure_callback 中不起作用

Question

问题:on_failure_callback 中使用的运算符中的模板未呈现。

def report_failure(context):
    send_email = EmailOperator(
        task_id="email_failed",
        to=emailreceipients,
        subject="{{execution_date}}",
        html_content=get_email_body() # Which returns "Body: {{execution_date}}"
    )

    # One solution I tried fails on: AttributeError: 'EmailOperator' object has no attribute 'render_template_fields'
    # Which is weird as this method is present on EmailOperator's base: BaseOperator
    # send_email.render_template_fields(context)

    send_email.execute(context)

default_args = {
    'on_failure_callback': report_failure
}

发送的电子邮件实际上包含 {{execution_date}} 而不是呈现的值。

在这个简单的情况下，我可以使用 .format(**context) 就地格式化这些模板化字符串，但我在 context 的其他地方重复使用此电子邮件发送不可用，并且模板工作正常。

更大的目标是在 DAG(或其子DAG)的任何步骤失败时触发此“错误处理程序”。

Answer 1

经过多次尝试不同的方法后，我发现以下解决方案有效:

解决方案A:不使用on_failure_callback，而是使用trigger_rule='all_failed'/'one_failed;创建任务

这是 Jacob 在 Airflow 的 Slack 上提出的建议，而且看起来效果很好。恕我直言，它在概念上是最简单的。

status_failed = SimpleHttpOperator(
    trigger_rule='all_failed', # See https://airflow.apache.org/docs/stable/concepts.html
    task_id='updateStatus',
    ...
)
email_failed = EmailOperator(
    trigger_rule='all_failed',
)

start_task >> do_the_thing >> status_success >> email_success
# Handling errors in case any job fails
email_success >> email_failed
email_success >> status_failed

示例:成功运行，将整个 DAG 的状态设置为“成功”，并将错误处理程序设置为跳过。

解决方案 B:使用 on_failure_callback 并手动渲染模板

正如阿什所说above 模板在那里不起作用的原因是模板不是由execute()渲染的，而是事先渲染的。我发现以下解决方案有效:

def report_failure(context):
    send_email = EmailOperator(
        task_id="email_failed",
        start_date=datetime(2015, 12, 1), # Any date in the past, if you won't set it you will get an error
        to=emailreceipients,
        subject="{{execution_date}}",
        html_content=get_email_body() # Which returns "Body: {{execution_date}}"
    )

    # Set DAG, otherwise we will get errors
    send_email.dag = context['dag']

    # Manually render templates
    # send_email.render_template_fields(context) # Working in Airflow 1.10.6
    # send_email.html_content = send_email.render_template('', send_email.html_content, context) # Working in Airflow 1.10.4
    # Looking at codebase seems to be working in both versions
    send_email.html_content = send_email.get_template_env().from_string(send_email.html_content).render(**context)

    send_email.execute(context)

问题在于这些功能似乎经常变化，因此可能很难更新。

解决方案 C:使用 context 变量格式化消息

正如雅各布建议的那样above

当生成模板的代码与不使用上下文的运算符(operator)共享时，它不能解决我的特殊情况。尽管如此，它对于更简单的情况可能会有所帮助，因此我将其发布在这里。