• Created by Unknown User (bolke), last updated by Unknown User (criccomini) on Jun 03, 2016 3 minute read

You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 5 Next »

There seems to be a bit of mysticism around the scheduler in Airflow, that makes people reluctant to review PRs in this area or contribute. That should not be the case. The scheduler is the core of Airflow it needs to be the most understood and readable bit of code. So here is a small write up to get you started.

A DAG consists of Tasks and obviously you need those tasks to run. To do that we schedule the dag in "schedule_dag" (jobs.py) to make it into a DagRun. This means that a DagRun is an instantiation of a DAG in time. Then we need to process the tasks (process_dag in jobs.py) of the current active dagruns. We simply loop over the available tasks and see if they are runnable (if ti.is_runnable()) . If a task is runnable it will get sent to the executor.

That's all there is to the basic concept. Really? Yes:

def _do_dags(self, dagbag, dags, tis_out):
    """
    Iterates over the dags and schedules and processes them
    """
    for dag in dags:
        self.logger.debug("Scheduling {}".format(dag.dag_id))
        dag = dagbag.get_dag(dag.dag_id)
        if not dag:
            continue
        try:
            self.schedule_dag(dag)
            self.process_dag(dag, tis_out)
            self.manage_slas(dag)
        except Exception as e:
            self.logger.exception(e)

Now it is a bit more messy a the moment. That is due to the fact that DagRuns are not yet first class citizens. In the past Airflow only had a notion of DagRun: it mentioned them in the code but they did not really exist, instead DAGs were instantiated into "Tasks in Time": TaskInstances. This creates architectural issues but also real life ones. Some of the issues in the current scheduler with "depend_on_past" stem from this, because a Task is not really able to answer "which task instance is the first in time?". It is also the reason why scheduler loops start to increase over time if DAGs get more complex (many tasks). The good news is work is underway to improve this (see the roadmap).

States and race conditions in the scheduler

States are used in Airflow to understand what the different tasks and dagruns are doing. We currently know the following states.

NONE = None
QUEUED = "queued"
RUNNING = "running"
SUCCESS = "success"
SHUTDOWN = "shutdown"  # External request to shut down
FAILED = "failed"
UP_FOR_RETRY = "up_for_retry"
UPSTREAM_FAILED = "upstream_failed"
SKIPPED = "skipped"

The scheduler processes tasks that have a state of NONE, QUEUED, and UP_FOR_RETRY. NONE is a newly created TaskInstance, QUEUED is a task that is waiting for a slot in an executor and UP_FOR_RETRY means a task that failed before but needs to be retried. At the moment the scheduler will not change the state of a Task. This creates the possibility of a race condition: if the executor cannot execute a task quickly enough its state will not be updated and it can get scheduled again. Due to the loop times being quite large in complex cases, this won't occur too often. As people try to limit scheduler loop times, or just prefer LocalExecutor more than the CeleryExecutor, they will run multiple schedulers at the same time, which increases the chances of this happening.

For this reason a "SCHEDULED" state is proposed in one of the PRs. This will not fully close the door to this race condition due to the fact that TaskInstance evaluate their own state in case of UP_FOR_RETRY (ie. checking if they should run now). Normally one would move this check to the scheduler was it not for that fact that we have backfills...

Pools

All tasks used to start with State.NONE. Unknown User (bolke)'s latest PR (AIRFLOW-128) introduces State.SCHEDULED when the TI has been created by the scheduler but not yet run. When the executor loads a TI, it eventually calls TI.run() which does one of two things:

  1. If the TI can be run, sets State.RUNNING and calls the execute() method.
  2. If the TI has a pool, sets State.QUEUED and returns

The scheduler has a prioritize_queued method which loads up all the queued tasks and tries to run them if there are slots available in their respective pools. That second run is the one that actually moves pooled tasks to State.RUNNING.

Backfills

Backfills are a bit of an awkward duck in the pond. They do not know about DagRuns, won't create them, and don't keep to the schedule so they can break "depend_on_past". They execute outside the scheduler and can therefore oversubscribe workers (using more resources than assigned). Backfills just create TaskInstances and start running them. In order to fix the scheduler and the race condition, first the scheduler and the backfills need to become aware of each other. This will make depend_on_past work and keep things in a consistent state. Avoiding oversubscribing the backfills should be managed by the scheduler.

 

 

 

  • No labels