如何递归解析传入Airflow DAG参数的{{ ds }}宏

问题场景 #

我们有一个Airflow DAG用于从数据库提取数据，日常按@daily调度，提取逻辑日期{{ ds }}当天的数据，SQL条件为：

WHERE date BETWEEN '{{ ds }}' AND '{{ ds }}'

日常调度运行正常。

执行回填时，希望通过Airflow UI的Trigger DAG w/ config手动触发，传入起始日期参数，将SQL改为：

WHERE date BETWEEN '{{ params.start_date }}' AND '{{ ds }}'

以此提取指定时间范围的数据。

为了让SQL同时适配日常单天提取和回填场景，我们设置默认参数：

{
   "start_date": "{{ ds }}",
   "start_date_alt": "ds"
}

手动传入静态日期（如2023-01-01）时，BashOperator输出符合预期：

Extracting data from 2023-01-01 2023-01-01 to 2023-10-28

但日常调度运行时，输出不符合预期：

Extracting data from {{ ds }} ds to 2023-10-27

我们需要输出实际日期，比如：

Extracting data from 2023-10-28 2023-10-28 to 2023-10-28

问题根源是{{ params.start_date }}返回的是Jinja模板字符串，但未被递归解析。尝试移除start_date_alt的花括号也无效。实际场景中我们使用GlueJobOperator，以下是示例DAG代码：

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator

DEFAULT_ARGS = {
    "start_date": datetime(2023, 9, 1),
}

with DAG(
    dag_id=f"recursive-template",
    default_args=DEFAULT_ARGS,
    schedule_interval="@daily",
    params={
        "start_date": "{{ ds }}",
        "start_date_alt": "ds"
    }
) as dag:
    bash_task = BashOperator(
        task_id="echo_dates",
        bash_command="echo Extracting data from {{ params.start_date }} {{ params.start_date_alt }} to {{ ds }}"
    )

解决方案 #

方法1：使用Jinja的template过滤器递归解析 #

Airflow允许在Jinja模板中使用template过滤器，对参数值进行二次解析。修改Operator的模板字符串：

bash_task = BashOperator(
    task_id="echo_dates",
    bash_command="echo Extracting data from {{ params.start_date | template }} {{ params.start_date_alt | template }} to {{ ds }}"
)

这个过滤器会把参数中的Jinja表达式（如{{ ds }}）重新解析为实际日期值，不管是默认参数还是手动传入的静态值都能正常工作。

方法2：在Operator中手动解析模板 #

如果使用的Operator不支持Jinja过滤器（比如部分自定义Operator），可以通过Airflow的get_template_context方法手动解析：

from airflow.templates import get_template_context

def resolve_template_value(value, context):
    template_env = context['dag'].get_template_env()
    return template_env.from_string(value).render(context)

# 在Operator的参数中使用（以BashOperator为例）
bash_task = BashOperator(
    task_id="echo_dates",
    bash_command=lambda context: f"echo Extracting data from {resolve_template_value(context['params']['start_date'], context)} {resolve_template_value(context['params']['start_date_alt'], context)} to {context['ds']}"
)

这种方式通过Lambda函数获取上下文，手动调用模板环境解析参数值，适合更复杂的场景。

方法3：针对GlueJobOperator的适配 #

对于GlueJobOperator，通常通过script_args传递参数，同样可以使用template过滤器：

from airflow.providers.amazon.aws.operators.glue import GlueJobOperator

glue_task = GlueJobOperator(
    task_id="run_glue_job",
    job_name="your-glue-job",
    script_args={
        "--start_date": "{{ params.start_date | template }}",
        "--end_date": "{{ ds }}"
    }
)

这样Glue作业接收到的--start_date参数会是解析后的实际日期，无论是日常调度还是手动回填都能正确处理。

注意事项 #

• 确保Airflow版本支持template过滤器：该功能在Airflow 2.x版本中可用，若使用1.x版本可能需要升级或使用手动解析的方式。
• 手动传入参数时，直接传递静态日期字符串（如"2023-01-01"）即可，不需要包裹Jinja表达式，template过滤器会直接返回原字符串。

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