Airflow中处理{{ts_nodash}}变量遇格式错误，求正确访问方法

问题原因 #

你的代码报错核心原因是：DAG解析阶段（Airflow加载DAG文件时），Python代码会优先执行，此时"{{ ts_nodash }}"只是一个普通字符串，还没被Airflow的模板引擎渲染成实际的时间戳。所以datetime.strptime尝试解析"{{ ts_nodash }}"这个字面量，自然匹配不上%Y%m%dT%H%M%S格式，抛出ValueError。

可行解决方案 #

方案1：直接在Jinja模板中完成时间计算 #

利用Airflow的Jinja模板支持，把时间计算逻辑直接写到args的模板字符串里（GKE Operator的args默认属于可模板化字段）。通过Airflow内置的macros工具类处理时间运算：

get_numbers = gke_wrapper.execute_gke_operator(
    task_id=f"get_numbers_{ctype_name}",
    labels=get_cost_labels(
        pod=Pod.pod,
        service="service",
        id="mmoe",
        environment="dev",
        component="airflow",
    ),
    cmd=["sh"],
    args=[
        "get_numbers.sh",
        output_table_name,
        dump_table_name,
        model_path,
        model_path,
        ctype_name,
        config.variants[ctype_name],
        # 直接在模板中计算：execution_date对应ts_nodash的原始时间，减去指定小时后格式化
        "{{ (execution_date - macros.timedelta(hours=" + str(config.delay_interval) + ")).strftime('%Y%m%dT%H%M%S') }}"
    ],
    nodepool=_NODE_POOL,
    image=_IMAGE,
    resource=_RESOURCES,
)

注意：如果config.delay_interval是动态变量，需要确保它能在DAG解析阶段被解析为字符串，或者将其定义为Airflow变量，通过{{ var.value.delay_interval }}在模板中引用。

方案2：用PythonOperator预处理时间，通过XCom传递 #

先创建一个Python任务计算目标时间戳，再把结果通过XCom传递给GKE任务：

from airflow.operators.python import PythonOperator

def calculate_target_ts(**context):
    # 从上下文获取execution_date，对应ts_nodash的原始时间
    execution_date = context["execution_date"]
    target_ts = (execution_date - timedelta(hours=config.delay_interval)).strftime("%Y%m%dT%H%M%S")
    # 将结果存入XCom
    return target_ts

# 预处理时间的Python任务
calc_ts_task = PythonOperator(
    task_id=f"calc_target_ts_{ctype_name}",
    python_callable=calculate_target_ts,
    provide_context=True
)

# GKE任务从XCom获取预处理后的时间戳
get_numbers = gke_wrapper.execute_gke_operator(
    task_id=f"get_numbers_{ctype_name}",
    labels=get_cost_labels(
        pod=Pod.pod,
        service="service",
        id="mmoe",
        environment="dev",
        component="airflow",
    ),
    cmd=["sh"],
    args=[
        "get_numbers.sh",
        output_table_name,
        dump_table_name,
        model_path,
        model_path,
        ctype_name,
        config.variants[ctype_name],
        # 从XCom获取结果
        "{{ ti.xcom_pull(task_ids='calc_target_ts_" + ctype_name + "') }}"
    ],
    nodepool=_NODE_POOL,
    image=_IMAGE,
    resource=_RESOURCES,
)

# 设置任务依赖
calc_ts_task >> get_numbers

方案3：自定义模板过滤器（适合重复使用场景） #

如果这个时间计算逻辑需要在多个任务中复用，可以自定义一个Jinja模板过滤器：

1. 在Airflow的plugins目录下创建过滤器文件（比如custom_filters.py）：

from datetime import timedelta

def subtract_hours(execution_date, hours):
    return (execution_date - timedelta(hours=hours)).strftime("%Y%m%dT%H%M%S")

# 注册过滤器
class CustomFiltersPlugin:
    name = "custom_filters"
    jinja_filters = {
        "subtract_hours": subtract_hours
    }

2. 在DAG中直接使用自定义过滤器：

args=[
    "get_numbers.sh",
    output_table_name,
    dump_table_name,
    model_path,
    model_path,
    ctype_name,
    config.variants[ctype_name],
    "{{ execution_date | subtract_hours(" + str(config.delay_interval) + ") }}"
]

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