Airflow如何多次触发DAG并让DAGRun排队逐个执行

方法一：利用Airflow内置DAG并发控制（最简方案） #

直接通过DAG配置限制同时运行的实例数，让Airflow自动处理触发请求的排队：

1. 核心配置
   在DAG定义中设置max_active_runs=1，调度器会自动将新触发的DAGRun标记为queued状态，仅当前一个DAGRun完全执行完毕后，下一个才会启动。所有触发请求都会生成独立的DAGRun，不会遗漏。
   示例代码：

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'serial_k8s_task_dag',
    default_args=default_args,
    max_active_runs=1,  # 关键：限制同时运行的DAGRun数量为1
    schedule_interval=None,  # 按需触发，禁用自动调度
    catchup=False,
) as dag:
    serial_task = KubernetesPodOperator(
        task_id='execute_serial_task',
        name='serial-k8s-task',
        image='your-custom-image:latest',
        cmds=['/path/to/your/command'],
        arguments=['--param1', 'value1'],
        is_delete_operator_pod=True,
        namespace='airflow-workloads',
    )

2. 触发方式
   无论是Airflow UI手动触发、REST API调用触发，还是外部系统触发，每次请求都会生成一个DAGRun并进入排队序列，按触发顺序逐个执行。

方法二：自定义队列+Sensor（复杂场景适配） #

如果需要自定义优先级、对接外部系统队列等更灵活的控制，可以结合外部队列（如Redis）与Airflow Sensor实现：

1. 触发入队逻辑
   每次触发时，将任务参数（镜像标签、命令、参数等）写入外部队列。可以通过触发时传递参数，或在DAG启动阶段用PythonOperator完成入队。

2. DAG监听执行
   DAG中仅保留一个循环监听队列的Sensor任务，检测到队列有任务时，取出参数并执行KubernetesPodOperator，完成后继续监听队列。
   示例代码（简化版）：

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.sensors.python import PythonSensor
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import redis
import json

def check_task_queue(**context):
    r = redis.Redis(host='redis-service', port=6379, db=0)
    return r.llen('serial_task_queue') > 0  # 检查队列是否有任务

def run_queued_task(**context):
    r = redis.Redis(host='redis-service', port=6379, db=0)
    task_params = json.loads(r.lpop('serial_task_queue'))
    
    k8s_task = KubernetesPodOperator(
        task_id='dynamic_k8s_execution',
        name='dynamic-serial-task',
        image=task_params['image'],
        cmds=task_params['cmds'],
        arguments=task_params['args'],
        is_delete_operator_pod=True,
        namespace='airflow-workloads',
        do_xcom_push=False
    )
    k8s_task.execute(context)

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
}

with DAG(
    'queue_driven_serial_dag',
    default_args=default_args,
    max_active_runs=1,
    schedule_interval=None,
    catchup=False,
) as dag:
    queue_sensor = PythonSensor(
        task_id='watch_task_queue',
        python_callable=check_task_queue,
        poke_interval=30,  # 每30秒检查一次队列
        mode='reschedule',  # 避免长期占用Worker资源
    )

    execute_task = PythonOperator(
        task_id='run_queued_k8s_task',
        python_callable=run_queued_task,
    )

    queue_sensor >> execute_task >> queue_sensor  # 形成循环监听逻辑

关键注意事项 #

• 任务幂等性：确保Kubernetes任务具备幂等性，重复执行不会产生异常结果或数据污染。
• 失败处理：配置retries和retry_delay参数，避免单个任务失败导致队列阻塞。
• 资源清理：KubernetesPodOperator需设置is_delete_operator_pod=True，执行完毕后自动销毁Pod，避免资源泄漏。

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