1. 首先确保 max_active_runs 参数已经正确设置。可以在 DAG 中指定 max_active_runs 或者在 Airflow 的配置文件中将该参数设置为一个默认值。例如，在 DAG 文件中设置 max_active_runs：

from airflow import DAG

default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2021, 1, 1), 'email': ['airflow@example.com'], 'email_on_failure': False, 'email_on_retry': False }

dag = DAG( 'example_dag', default_args=default_args, description='Example DAG', schedule_interval=timedelta(days=1), max_active_runs=1 )

2. 如果 DAG 配置正确，但是 max_active_runs 参数仍然不起作用，可能是由于 AWS 托管的 Airflow 的默认行为。为了解决这个问题，可以在 DAG 文件中添加一些代码来限制 DAG 运行次数：

from airflow.models import DAG, DagRun from airflow.operators.python_operator import PythonOperator

def check_max_active_runs(**kwargs): dag_id = kwargs['dag'].dag_id max_active_runs = kwargs['dag'].max_active_runs

active_runs = DagRun.find(dag_id=dag_id, state='running')
if len(active_runs) >= max_active_runs:
    return False

return True

dag = DAG( dag_id='example_dag', max_active_runs=1, schedule_interval=timedelta(days=1), )

check_max_active_runs_task = PythonOperator( task_id='check_max_active_runs', python_callable=check_max_active_runs, provide_context=True, dag=dag )

add other tasks here

Use DagRun.find() 函数获取正在运行的 DAG 实例，如果当前 DAG 实例数量超出最大值，就停止 DAG 运行。

这种方法可以确保 DAG 运行次数不超过指定数量。