Airflow模板未渲染问题：自定义TemplatedLivyOperator无法解析{{ ts }}

问题原因 #

你遇到的ValueError: Invalid isoformat string: '{{ ts }}'错误，本质是DAG解析阶段Python代码会优先执行，而Airflow的模板渲染（把{{ ts }}替换为实际执行日期）是在任务执行阶段才触发的。你当前的写法是直接把字符串'{{ ts }}'传给datetime.fromisoformat()，自然会因格式不合法报错。

解决方案 #

方案1：使用Jinja2内置日期过滤器直接格式化 #

Airflow的Jinja2模板支持strftime过滤器，直接在模板字符串内完成日期格式化，无需提前用Python处理：

op = TemplatedLivyOperator(
    dag=dag,
    task_id='test1',
    livy_conn_id='livy_default',
    name="run1_{{ ts | strftime('%Y-%m-%d-%H') }}",
    args=[substrate, env, "{{ ts | strftime('%Y-%m-%d-%H') }}"]
)

模板渲染时，{{ ts | strftime('%Y-%m-%d-%H') }}会被直接替换为格式化后的日期字符串，完美契合你的需求。

方案2：自定义Jinja2过滤器（适用于复杂日期逻辑） #

如果默认过滤器满足不了复杂处理需求，可以在DAG中注册自定义过滤器：

from datetime import datetime
from airflow import DAG

# 自定义日期格式化函数
def format_execution_ts(ts):
    return datetime.fromisoformat(ts).strftime('%Y-%m-%d-%H')

# 在DAG中注册自定义过滤器
dag = DAG(
    'your_dag_id',
    default_args=your_default_args,
    template_globals={'format_execution_ts': format_execution_ts}
)

# 使用自定义过滤器
op = TemplatedLivyOperator(
    dag=dag,
    task_id='test1',
    livy_conn_id='livy_default',
    name="run1_{{ format_execution_ts(ts) }}",
    args=[substrate, env, "{{ format_execution_ts(ts) }}"]
)

关键注意点 #

模板字段的渲染时机晚于DAG解析阶段，所有针对模板变量的处理逻辑必须放在Jinja2模板语法中，而不能在Python代码层面提前执行——后者会把模板标记（如{{ ts }}）当作普通字符串处理，导致报错。

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