如何为Airflow DAG配置默认值为逻辑日期的参数

你的代码问题在于DAG的params默认值不会自动触发Jinja模板渲染，所以"{{ ds }}"会被当作字符串直接输出。下面提供两种可行的解决方案：

方法1：在BashOperator的模板中使用默认值语法 #

直接在bash_command里用Jinja的默认值逻辑，判断参数是否存在，不存在则使用ds（对应logical_date的YYYY-MM-DD格式）：

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

# 定义DAG，无需在params中预设默认值
dag = DAG(
    dag_id="test",
    start_date=days_ago(1),
    schedule_interval="@daily",
    params={}  # 可留空，或仅定义参数名不设值
)

print_param_task = BashOperator(
    task_id="print_param",
    # 使用Jinja的get方法，参数不存在时返回ds作为默认值
    bash_command='echo "{{ params.get(''date_param'', ds) }}"',
    dag=dag
)

触发逻辑：

• 手动传入date_param参数时，打印传入值
• 未传入参数时，自动使用当前DAG运行的logical_date对应的ds值

方法2：用PythonOperator实现更灵活的逻辑 #

如果需要对logical_date做更复杂的格式化或处理，推荐使用PythonOperator，直接从上下文获取logical_date对象：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import days_ago

def print_date_param(**context):
    # 从上下文获取参数，无参数时用logical_date格式化后的字符串
    date_param = context["params"].get(
        "date_param", 
        context["logical_date"].strftime("%Y-%m-%d")
    )
    print(f"当前使用的日期参数：{date_param}")

dag = DAG(
    dag_id="test",
    start_date=days_ago(1),
    schedule_interval="@daily",
)

print_param_task = PythonOperator(
    task_id="print_param",
    python_callable=print_date_param,
    provide_context=True,  # 必须开启才能获取上下文变量
    dag=dag
)

这种方式可以直接操作logical_date这个datetime对象，根据需求调整格式（比如%Y%m%d），扩展性更强。

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