A reproduction repo for a Scheduling bug in AirFlow 2.2.3

Last update: Feb 09, 2022

Related tags

Overview

How to run

docker-compose build
docker-compose up

Setup

Have 3 DAGs:

~240 Tasks, executes every hour, runs for about 45-50 minutes total
~600 Tasks, executes every 5 days, runs for days
~10 Tasks, executes on trigger, runs for 10-50 minutes

DAGs run on the default_pool with max_active_tasks set to 2.

My AirFlow config file has the following:

parallelism = 32
default_pool_task_slot_count = 6
executor = LocalExecutor
default_task_weight_rule = absolute

parallelism was set initially like that. default_pool_task_slot_count was set to 6, because it seemed rational that if all of my 3 dags are executing at the same time, the maximum amount of tasks that can be executed is 3*2=6.

The problem:

Almost every time any one of the DAGs is executed, no tasks from other DAGs will start until all the tasks from the first one finish. That is, if slow DAG_2 with 600 tasks starts, DAG_1 will have to wait for days to start even a single Task.

The Logs for the Scheduler look like this:

scheduler_1  | 
scheduler_1  | [2022-02-09 10:02:27,116] {scheduler_job.py:288} INFO - 4 tasks up for execution:
scheduler_1  |  
   
    
scheduler_1  |  
    
     
scheduler_1  |  
     
      
scheduler_1  |  
      
       
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:322} INFO - Figuring out tasks to run in Pool(name=default_pool) with 4 open slots and 4 task instances ready to be queued
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:357} INFO - Not executing 
       
         since the number of tasks running or queued from DAG Slow_DAG is >= to the DAG's max_active_tasks limit of 2 scheduler_1 | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks scheduler_1 | [2022-02-09 10:02:27,119] {scheduler_job.py:357} INFO - Not executing 
        
          since the number of tasks running or queued from DAG Slow_DAG is >= to the DAG's max_active_tasks limit of 2 scheduler_1 | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks scheduler_1 | [2022-02-09 10:02:27,120] {scheduler_job.py:357} INFO - Not executing 
         
           since the number of tasks running or queued from DAG Slow_DAG is >= to the DAG's max_active_tasks limit of 2 scheduler_1 | [2022-02-09 10:02:27,120] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks scheduler_1 | [2022-02-09 10:02:27,120] {scheduler_job.py:357} INFO - Not executing 
          
            since the number of tasks running or queued from DAG Slow_DAG is >= to the DAG's max_active_tasks limit of 2 scheduler_1 | [2022-02-09 10:02:27,120] {scheduler_job.py:410} INFO - Setting the following tasks to queued state: scheduler_1 |

I have fixed this by setting

parallelism = 1000
default_pool_task_slot_count = 999

The reason why this occurs and why the solution works

We need to look at this method of the Scheduler: https://github.com/apache/airflow/blob/2.2.3/airflow/jobs/scheduler_job.py#L229
In it, the overall logic is the following:

Receive max_tis(== parallelism - active running tasks from the config) in the arguments. In my case that would be 30
Calculate how many free slots in all all Pools we have. In my case that would be 4.
Select the minimum and update the max_tis variable: max_tis = min(4, 30) = 4
Query the DB for tasks that are Scheduled in an unpaused DAGs that are running normally and order them by DAG execution date.
Limit the query by max_tis.
Loop over each returned task and check if we can run it. Run, if possible.

Now, since the tasks are ordered by execution date, this will return us tasks for a DAG that started first, which lets say was DAG_2. It has over 600 tasks. The LIMIT operation will return 4 of them. Each off these tasks cannot be run since there are already 2 tasks running and the max_active_tasks of the DAG is 2. Thus, we just wait till one of these tasks finish and start another task from the same DAG.

A reproduction repo for a Scheduling bug in AirFlow 2.2.3

Related tags

Overview

How to run

Setup

The problem:

The reason why this occurs and why the solution works

Owner

Ilya Strelnikov

Just a little benchmark for scrapper PC's

An After Effects render queue for ShotGrid Toolkit.

RISE allows you to instantly turn your Jupyter Notebooks into a slideshow

Snek-test - An operating system kernel made in python and assembly

Install Firefox from Mozilla.org easily, complete with .desktop file creation.

Fisherman is a free open source fishing bot written in python.

Solves Maths24 problems for you!

Winxp_python3.6.15 - Python 3.6.15 For Windows XP SP3

📜Generate poetry with gcc diagnostics

Aplicação que envia regularmente um email ao utilizador com todos os filmes disponíveis no cartaz dos cinemas Nos.

combs is a package used to generate all possible combinations of a given length k on a given set.

Zues Auto Claimer Leaked By bazooka#0001

Recreate the joys of Office Assistant from the comfort of the Python interpreter

TickerRain is an open-source web app that stores and analysis Reddit posts in a transparent and semi-interactive manner.

Simple calculator made in python

NYCU(NCTU)-差勤-助教

Islam - This is a simple python script.In this script I have written all the suras of Al Quran. As a result, by using this script, you can know the number of any sura at the moment.

PressurePlate is a multi-agent environment that requires agents to cooperate during the traversal of a gridworld.

Python wrapper around Apple App Store Api

Windows Task Manager with special features, written in Python.