Airflow DAG参数默认逻辑日期设置教程

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

dag = DAG(
    dag_id="test_dag_params_issue",
    start_date=days_ago(1),
    schedule_interval="@daily",
    params={"date_param": "{{ ds }}" } # 这里的{{ ds }}会被当作字符串字面量
)

print_param_task = BashOperator(
    task_id="print_param",
    bash_command='echo "参数值: {{ params.date_param }}"',
    dag=dag
)

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

# 定义一个独特的虚拟默认值，以避免与实际传入的参数冲突
DUMMY_DEFAULT_VALUE = "AIRFLOW_DEFAULT_LOGICAL_DATE_PLACEHOLDER"

with DAG(
    dag_id="airflow_default_logical_date_param",
    start_date=days_ago(1),
    schedule_interval="@daily",
    catchup=False,
    # 在params中设置一个虚拟的默认值
    params={"date_param": DUMMY_DEFAULT_VALUE }
) as dag:

    # 定义BashOperator任务
    # 在bash_command中利用Jinja条件判断来决定参数的最终值
    print_param_task = BashOperator(
        task_id="print_param",
        bash_command=f'echo "当前逻辑日期: {{ ds }}" && '
                     f'echo "传入或默认日期参数: {{ ds if params.date_param == "{DUMMY_DEFAULT_VALUE}" else params.date_param}}"',
        dag=dag
    )

    # 另一个示例：使用PythonOperator
    from airflow.operators.python import PythonOperator

    def _process_date_param(**kwargs):
        ti = kwargs['ti']
        # 从task_instance中获取经过Jinja渲染后的参数
        rendered_date_param = ti.xcom_pull(task_ids=None, key='rendered_date_param') # 假设BashOperator将它推送到XCom
        # 或者更直接地，如果PythonOperator的op_kwargs是可模板化的
        # 在PythonOperator中直接访问模板化参数通常需要通过 op_kwargs 或 context
        # 这里为了演示，我们假设将Jinja表达式直接放在op_kwargs中
        date_param_from_context = kwargs['params'].get('date_param')
        if date_param_from_context == DUMMY_DEFAULT_VALUE:
            final_date = kwargs['ds'] # 直接使用上下文中的ds
        else:
            final_date = date_param_from_context
        print(f"Python任务处理的日期参数: {final_date}")

    python_task = PythonOperator(
        task_id="python_process_param",
        python_callable=_process_date_param,
        # op_kwargs通常是可模板化的，但直接在这里使用Jinja表达式会更复杂
        # 推荐在Python函数内部根据上下文判断
        provide_context=True, # 确保上下文（包括ds）被传入
        dag=dag
    )

    # 任务依赖
    print_param_task >> python_task