Airflow如何通过Jinja模板按DAG标签设置Operator的retries参数

错误原因 #

• Airflow 1.10 原生PythonOperator的默认模板渲染字段列表不包含retries，直接给retries传入Jinja模板字符串不会被解析，只会被当作普通字符串处理，因此无法生效。
• 你写的模板逻辑本身没有问题，只需要让retries字段支持模板渲染，就可以正常读取任务上下文里的dag对象属性。

实现方案 #

不需要修改任何现有DAG文件，仅修改common.py即可完成需求适配：

步骤1：自定义支持retries模板渲染的Operator子类 #

继承原生PythonOperator，将retries加入模板渲染字段列表：

from airflow.operators.python_operator import PythonOperator

class RetryTemplatedPythonOperator(PythonOperator):
    # 扩展原生模板字段，新增retries支持Jinja渲染
    template_fields = PythonOperator.template_fields + ('retries',)

步骤2：替换tasks函数中的Operator实现 #

将原来的PythonOperator替换为自定义子类，修正模板引号转义问题：

def foo(job_name):
    raise Exception()

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

补充说明 #

• 该方案完全兼容现有DAG的调用逻辑，不需要调整任何dag.py文件的代码，适配存量代码体量大的场景。
• Airflow 1.10的任务运行上下文默认包含dag对象，模板中可以直接读取dag.tags属性，无需额外注入参数。
• 如果需要调整标签匹配规则、重试次数，仅修改模板中的判断逻辑即可。

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