如何在Airflow中确保一个DAG执行完毕后再启动下一次运行？

需要将数据迁移至另一数据库，使用Airflow运行由ETL流程组成的DAG，每次运行处理一天的数据，需补追三年的历史数据。核心需求是必须等待前一次DAG执行完毕后再启动下一次运行，但每次运行的执行时间未知。

尝试设置最小schedule_interval并将max_active_runs设为1后，出现大量日期被跳过的问题：Airflow会在schedule_interval结束时尝试触发新的DAG运行，虽然因max_active_runs=1限制不会实际执行，但补追日期的变量已经递增，导致部分日期未被处理。

DAG代码如下：

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

def launch_date():
    # 自定义日期生成逻辑
    pass

def launch_flow(date):
    # ETL流程逻辑
    pass

with DAG(
        dag_id='catch_up',
        default_args={
            'owner': 'airflow',
            'start_date': datetime.now(),
            'depends_on_past': False,
            'retries': 1,
        },
        description=' ',
        schedule_interval='*/2 * * * *',
        max_active_runs=1,
        catchup=False

) as dag: 

    date = launch_date()
    start_etl = PythonOperator(
            task_id='flow',
            python_callable=launch_flow,
            op_args=[date]
    )

    ...

解决方案 #

1. 利用Airflow原生特性实现串行执行 #

• 将depends_on_past设为True：强制当前DAG运行必须等待上一次运行成功完成后才会触发。
• 修正start_date：设置为补追历史数据的起始日期（如三年前的某天datetime(2021, 1, 1)），而非datetime.now()。
• 开启catchup=True：Airflow会自动生成从start_date到当前日期的所有历史运行任务，结合max_active_runs=1，实现严格串行处理。

修改后的核心DAG参数：

default_args={
    'owner': 'airflow',
    'start_date': datetime(2021, 1, 1),
    'depends_on_past': True,
    'retries': 1,
},
schedule_interval='@daily',
max_active_runs=1,
catchup=True

2. 用Airflow内置执行日期传递处理日期 #

放弃自定义launch_date()生成日期的逻辑，直接使用Airflow的execution_date作为每日处理的日期，避免手动维护日期导致的跳过问题。修改PythonOperator：

start_etl = PythonOperator(
    task_id='flow',
    python_callable=launch_flow,
    op_args=[{{ execution_date.strftime('%Y-%m-%d') }}],
    provide_context=True
)

对应的launch_flow函数直接接收该日期参数即可。

3. 避免短间隔调度干扰 #

之前使用*/2 * * * *（每2分钟）的短间隔会导致Airflow频繁触发新运行请求，即使被max_active_runs拦截，也可能干扰日期逻辑。既然是按天处理数据，直接使用@daily调度间隔更合理，配合catchup=True就能自动生成所有历史日期的运行任务。

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