如何触发指定日期的DAG并让跨DAG任务沿用该日期？

针对你遇到的问题——触发主DAG时指定了2022-09-20，但其中属于另一个DAG的run_biopackaging任务默认用当日日期执行，以下是几种可行的实现方案：

方案1：用TriggerDagRunOperator传递执行日期 #

这是最直接的跨DAG触发方式，通过该Operator将主DAG的触发日期传递给目标DAG，让目标DAG以指定日期运行。

from airflow.operators.trigger_dagrun import TriggerDagRunOperator

trigger_biopackaging_dag = TriggerDagRunOperator(
    task_id="trigger_biopackaging_dag",
    trigger_dag_id="biopackaging_dag",  # 替换为目标DAG的实际ID
    execution_date="{{ execution_date }}",  # 传递主DAG的执行日期
    reset_dag_run=True,  # 若目标DAG已有该日期的运行记录，允许重置
    wait_for_completion=False,  # 根据业务需求选择是否等待目标DAG完成
    dag=dag  # 关联到当前主DAG
)

配置后，目标DAG会以2022-09-20作为执行日期启动，其内部的run_biopackaging任务自然会沿用这个日期，而非当日日期。

方案2：调整ExternalTaskSensor的日期匹配逻辑 #

如果你的主DAG是通过ExternalTaskSensor等待run_biopackaging任务完成，需要修改传感器的execution_date_fn参数，让它指向主DAG的触发日期：

from airflow.sensors.external_task import ExternalTaskSensor

wait_biopackaging_task = ExternalTaskSensor(
    task_id="wait_biopackaging_task",
    external_dag_id="biopackaging_dag",
    external_task_id="run_biopackaging",
    execution_date_fn=lambda dt: dt,  # 使用当前DAG的执行日期匹配目标任务
    mode="poke",
    dag=dag
)

注意：这种方式需要确保目标DAG已经在2022-09-20日期有对应的运行，通常需要配合方案1一起使用，先触发目标DAG在指定日期运行，再用传感器等待其完成。

方案3：直接复用任务逻辑并传递日期参数 #

如果run_biopackaging的逻辑可以从目标DAG中剥离，你可以在主DAG中通过PythonOperator直接调用该逻辑，并将触发日期作为参数传入：

from airflow.operators.python import PythonOperator

def execute_biopackaging(execution_date, **context):
    # 这里复制原run_biopackaging任务的执行逻辑
    # 所有需要使用日期的地方，都用传入的execution_date
    process_data(date=execution_date)

run_biopackaging = PythonOperator(
    task_id="run_biopackaging",
    python_callable=execute_biopackaging,
    op_kwargs={"execution_date": "{{ execution_date }}"},
    provide_context=True,
    dag=dag
)

这种方式无需跨DAG触发，直接在主DAG内完成任务，完全复用指定的触发日期。

注意事项 #

• Airflow 2.x版本中，execution_date被data_interval_start替代，若使用新版本，需将模板变量改为{{ data_interval_start }}。
• 确保目标DAG的start_date早于或等于2022-09-20，且catchup设置正确（若需要回溯运行）。
• 若目标DAG已有指定日期的运行记录，方案1中需设置reset_dag_run=True，否则触发会失败。

内容的提问来源于stack exchange，提问作者Sebastián Salas