Airflow DAG运行优先级：如何将特定排队DAG Run转为执行状态？

1. 利用Airflow原生DAG Run优先级参数（Airflow 2.2+） #

Airflow 2.2及后续版本支持直接为DAG Run设置全局优先级，这个参数会覆盖任务级的priority_weight，让整个DAG Run的所有任务统一使用该优先级调度。

• 手动触发场景：在UI触发DAG的页面，找到Priority Weight输入框，填入自定义数值（数值越大优先级越高），提交后该DAG Run会被调度器优先选中执行。
• 代码触发场景：调用DagRun.create()方法时传入priority_weight参数：

from airflow.models import DagRun
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunType
from datetime import datetime

DagRun.create(
    dag_id="your_target_dag",
    run_id=f"custom_run_{datetime.utcnow().timestamp()}",
    state=DagRunState.QUEUED,
    execution_date=datetime.utcnow(),
    priority_weight=100,  # 自定义优先级数值
    dag_run_type=DagRunType.MANUAL,
    conf={"customer_id": "target_customer"}  # 可附带客户标识用于后续识别
)

2. 自定义调度器插件动态调整优先级 #

如果需要根据客户标识自动识别并提升优先级，可以编写调度器插件，定时扫描排队中的DAG Run并修改其优先级：

from airflow.plugins_manager import AirflowPlugin
from airflow.schedulers.base_scheduler import BaseScheduler
from airflow.models import DagRun
from airflow.utils.state import DagRunState
from datetime import timedelta

class CustomerPrioritySchedulerPlugin(AirflowPlugin):
    name = "customer_priority_scheduler"

    @staticmethod
    def on_scheduler_start(scheduler: BaseScheduler):
        def adjust_customer_priority():
            # 筛选处于排队状态的DAG Run
            queued_runs = DagRun.find(state=DagRunState.QUEUED)
            for run in queued_runs:
                # 从DAG Run的配置中提取客户标识
                customer_id = run.conf.get("customer_id")
                if customer_id == "high_priority_customer":
                    run.priority_weight = 200  # 提升优先级
                    run.update()

        # 每隔30秒执行一次优先级检查调整
        scheduler.scheduler_job.add_interval_job(
            adjust_customer_priority,
            timedelta(seconds=30)
        )

将该插件文件放入Airflow的plugins目录，重启调度器即可生效。

3. 通过REST API手动干预排队的DAG Run #

如果需要临时手动提升特定排队DAG Run的优先级，可使用Airflow REST API修改其参数：

curl -X PATCH \
  http://your-airflow-webserver:8080/api/v1/dags/your_dag_id/dagRuns/target_run_id \
  -H "Content-Type: application/json" \
  -d '{"priority_weight": 150}'

调度器会在下一轮调度周期中优先处理该高优先级的DAG Run。

注意事项 #

• 确保Airflow版本不低于2.2，否则DAG Run级的priority_weight参数无法生效。
• 优先级数值无严格范围限制，调度器会优先调度数值更高的DAG Run。
• 自定义插件需确保调度器有足够权限修改DAG Run数据，且插件代码需符合Airflow插件规范。

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