[Airflow] task, dag 우선순위 설정 (priority_weight)

airlfow에서 많은 Dags를 운영중이면 당연히 많은 task가 실행될때!!

>> 각 task가 실행되며 각자의 우선순위에 따라 airflow가 제한된 리소스내에서 실행시킬 task 우선순위 전략을 짬

 

Airflow 전략 방법

- priority_weight와 weight_rule으로 나눌수 있으며

- 각 task는 priority_weight를 가지며, default로 1의 값을 가짐

- weight_rule은 downstream, upstream, absolute가 있으며, default는 downstream

 

weight_rule  

- downstream  : 가중치는 다운스트림 priority_weight 누적 합계로 계산 

- upstream  : 가중치는 업스트림 task의 priority_weight 누적합계로 계산  

- absolute : 가중치는 해당 task의 priority_weight로 계산

 

예시

- A라는 task가 실행중일때 A태스크의 하위 task의 priority_weight는 합한 sum 값을 기준으로 각 task의 우선순위가 매겨짐

  1. DAG TEST1 : a >> b >> c

  2. DAG TEST2 : a1 >> b1>> c1 >> d1 >> [e1, e2, e3]

 - a와 a1중 실행 우선순위가 높은것은?!

>> a1=6, a=2점으로 a1이 우선 실행됨

 

 

코드 샘플

import pendulum

default_args = {
	"owner": "user",
 	"start_date": pendulum.datetime(2024,01,01),
	"provide_context": True,
	"weight_rule": "absolute",
}

dag = DAG(
    'my_dag',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
)

 

task1 = PythonOperator(
    task_id='task1',
    python_callable=some_function,
    priority_weight=5,
    dag=dag,
)

 

참고

Priority Weights — Airflow Documentation

Understanding priority_weight in Apache Airflow - FAQ November 2023