使用过去的dagrun日期创建Airflow DAG成功。

要使用过去的dagrun日期创建Airflow DAG，可以通过以下步骤实现：

1. 首先，导入所需的模块和类：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator

2. 创建一个函数来定义DAG的结构和任务。在这个函数中，使用execution_date参数来获取DAG运行的日期，并将其减去一天以得到过去的日期：

def create_dag(dag_id, schedule, default_args):
    with DAG(dag_id, schedule_interval=schedule, default_args=default_args) as dag:
        past_execution_date = "{{ execution_date - macros.timedelta(days=1) }}"
        
        # 添加任务
        task1 = DummyOperator(task_id='task1', dag=dag, start_date=past_execution_date)
        task2 = DummyOperator(task_id='task2', dag=dag, start_date=past_execution_date)
        task3 = DummyOperator(task_id='task3', dag=dag, start_date=past_execution_date)
        
        # 设置任务之间的依赖关系
        task1 >> task2 >> task3
        
    return dag

在这个函数中，我们使用{{ execution_date - macros.timedelta(days=1) }}来获取过去的DAG运行日期。

3. 定义DAG的默认参数和调度间隔：

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2021, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}
schedule = timedelta(days=1)

在这里，我们将start_date设置为一个过去的日期，以确保DAG可以从过去的日期开始运行。

4. 调用create_dag函数来创建DAG：

dag_id = 'example_dag'
dag = create_dag(dag_id, schedule, default_args)

通过传递DAG的ID、调度间隔和默认参数，我们可以创建一个名为example_dag的DAG。

最后，将DAG添加到Airflow的DAG列表中：

globals()[dag_id] = dag

这样，我们就可以在Airflow中看到并运行这个DAG了。