Tracking:

https://github.com/airbnb/airflow/compare/master...jlowin:dagrun-refactor 

Introduction

...

By tweaking the DagRunJob, we can easily recreate the behavior of the current SchedulerJob and BackfillJob. The Scheduler simply runs forever and picks up ALL active DagRuns in collect_dagruns(); Backfill generates DagRuns corresponding to the requested start/end dates and submits them to itself prior to initiating its loop.

...

https://github.com/airbnb/airflow/compare/master...jlowin:dagrun-refactor 

Changes

• (major API changes are tracked in UPDATING.md)

• DagRuns

  • A DagRun represents the execution of a certain DAG on a date, just as a TaskInstance represents the execution of a certain task on a date. As such, we don't want multiple DagRuns for the same dag/date, because they would all just point at the same taskinstances and therefore have no additional value.

    • This appears to already be enforced in Airflow via unique constraint but needs to be fleshed out. Right now it doesn't create any issues because only the Scheduler creates DagRuns, and only in sequence, but if Backfill also created a DagRun it could create a conflict with an existing scheduled DagRun.

  If a DagRun uniquely identifies a dag/date, then a few current DagRun attributes become meaningless, in particular id, run_id, and external_trigger. For example, we don't need a run_id if (dag_id, execution_date) is sufficient to identify a DagRun. These fields would be helpful if more than one DagRun could point at a certain day, but that has never been allowed and is simply more formally enforced in Scheduler 2.0.

  • ** The conf parameter should probably be rethought as well, for the same reason. It's tied to the method of execution. **

  • DagRuns can be locked (and unlocked). DagRunJob locks DagRuns when it's trying to execute them, and won't try to execute DagRuns if they are locked. This means multiple DagRunJobs can all run at the same time without stepping on each other. In particular, a BackfillJob might create DagRuns that are actually executed by the Scheduler. This sort of cooperation is ok -- DagRunJobs don't care how their DagRuns get executed, just that they do.

  • DagRuns have a "run" method which loads up all the TaskInstances they cover and tries to execute them (one time).

• DAGs

  • schedule_dag() The logic for scheduling a DAG moves from the Scheduler to the DAG itself, where it can be more easily reused.

  • update_dagrun_states() The DAG can review its outstanding DagRuns and update their states (Pending -> Active, Active -> Success/Failed)

• Scheduler behavior

  • Scheduling logic

    1. if schedule_interval == @once and has never been scheduled, run the dag now

    2. PREVIOUSLY:if the DAG has never been scheduled, see if it's ever been run at all. If it has run, the first scheduled date is 4 schedule_intervals prior to that run. If it's never run, use the earliest task start date (but not the dag start date?)

       NOW: if the DAG has never been scheduled, figure out the first date that SHOULD have been scheduled (probably dag.start_date + schedule_interval)

    3. if the DAG has been scheduled, add schedule_interval to the last scheduled date
    4. make sure the next run date is >= the dag's start_date
    5. make sure the next run date PLUS schedule_interval is <= now, indicating that the entire period has passed

  • In addition to scheduling new DagRuns, Scheduler tries to run ANY active DagRun (possibly restricted to specific dag_ids). This means Scheduler will try to run backfills and subdags, if they have created DagRuns.

  • Scheduler used to prioritize ALL queued tasks. Now it only prioritizes tasks that correspond to its active DagRuns (which could potentially mean ALL, but not necessarily)

...