Airflow - 使用 Jinja 模板设置操作员的重试次数答案

【问题标题】：Airflow - Set retries of operator using Jinja templateAirflow - 使用 Jinja 模板设置操作员的重试次数
【发布时间】：2021-11-19 17:44:52
【问题描述】：

我有一组 Airflow DAG，每个 DAG 都有一组与之关联的标签。我还有一个 PythonOperator，我的所有 DAG 都在使用它。

如果 DAG 应用了特定标签，我需要设置 Operator 的 retries 参数，否则不需要。

代码如下所示：

dag.py

from common import tasks

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.helpers import chain

dag = DAG(
'test_retry_dag',
tags=['abc','xyz'],
default_args=default_args,
schedule_interval='* * * * *',
)
    

with dag:
    chain(*tasks())

common.py

def foo(job_name):
    raise Exception()

def tasks():
    return [PythonOperator(
        task_id='auto_retry',
        retries='{{5 if "abc" in  dag.tags else 0 }}',
        python_callable=foo
    )]

由于tasks 是在一个单独的模块中定义的，因此我无法访问那里的 DAG 对象（将其作为参数传递是不可行的，因为这将涉及在非常大的代码库中对每个 DAG 进行更改。因此，我感觉使用 Jinja 模板是可行的方法，但是上面的代码不起作用，我认为我使用模板的方式是错误的。

有人可以帮帮我吗？

注意：我使用的是 Airflow 1.10，目前无法更新到 Airflow 2

【问题讨论】：

标签： python jinja2 airflow

【解决方案1】：

如果您查看 BaseOperator 代码，那么您会发现创建一个运算符实例（在您的情况下为 PythonOperator）而不在构造函数中传递 dag 参数，从 settings.CONTEXT_MANAGER_DAG 获取其值。

https://github.com/apache/airflow/blob/1.10.15/airflow/models/baseoperator.py#L420-L421

鉴于您描述的任务方法的用法，它看起来总是使用上下文 dag。然后您可以更改它以执行以下操作。

from airflow import settings

def foo(job_name):
    raise Exception()

def tasks():
    dag = settings.CONTEXT_MANAGER_DAG
    return [PythonOperator(
        task_id='auto_retry',
        retries=5 if "abc" in dag.tags else 0,
        python_callable=foo
    )]

【讨论】：