如何为Airflow中同一DAG的多个运行实例设置优先级

Airflow原生支持为单个DAG运行实例（DAG Run）分配独立优先级权重，优先级高于DAG/任务级别的默认权重配置，完全可以满足你的需求，常用实现方式如下：

1. 手动/半自动触发场景直接指定优先级 #

Airflow 2.3及以上版本已原生支持在创建DAG Run时传入priority_weight参数，调度器会优先调度权重数值更大的待运行实例：

• UI触发：在DAG触发弹窗的「优先级权重」输入框填入高于其他等待实例的数值即可
• CLI触发：执行命令时添加参数，示例：
  airflow dags trigger --priority-weight 200 your_dag_id
• API触发：在POST请求的body中添加"priority_weight": 200字段

你场景中只需要给第10个运行实例设置高于其余13个等待实例的权重，调度器就会优先将它分配到空闲执行槽位。

2. 批量自动触发场景动态分配优先级 #

如果你是批量触发30个运行实例，不想手动逐个设置权重，可以直接在DAG定义代码中通过动态逻辑给特定实例分配高优先级：

from airflow import DAG
from datetime import datetime

def calc_priority_weight(dag):
    # 可根据conf参数、执行时间等维度判断是否为需要优先的第10个实例
    current_dagrun = dag.get_last_dagrun()
    if not current_dagrun:
        return 10
    # 示例：触发时传入的conf中带instance_seq标识实例序号，等于10时分配高权重
    if current_dagrun.conf.get("instance_seq", 0) == 10:
        return 200
    return 10

with DAG(
    dag_id="your_batch_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    # priority_weight支持传入可调用对象，创建DAG Run时自动计算权重
    priority_weight=calc_priority_weight
) as dag:
    # 你的任务定义
    ...

注意点 #

• 权重值仅做相对比较，不需要设置过大数值，只需保证目标实例权重高于其余等待实例即可
• 如果你需要同时调整目标实例内部的任务优先级，可以在任务定义时用相同的动态逻辑设置任务级别的priority_weight参数
• 如果你仍在使用Airflow 1.x版本，需要通过自定义调度器扩展实现该能力，建议优先升级到2.x稳定版本

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