...

• Promote DagRun as a first-class Airflow object
  • Remove attributes that didn't makes sense, like external_trigger -- in a pragmatic sense, we don't really care, and the uniqueness constraint on (DAG, execution_date) means this is irrelevent anyway
  • DagRuns can manage their own state
  • DagRuns can be locked or unlocked by a specific lock_id, which guarantees that only one job is trying to execute the DagRun at a time
• Create a DagRunJob which is used to manage the execution of one or more DagRuns
• Implement both SchedulerJob and BackfillJob as subclasses of DagRunJob
  • The only difference is how they acquire DagRuns (Scheduler queries for all active DagRuns, BackfillJob works through a pre-specified list)
• Move much of the existing logic for figuring out the next schdeuled run from the Scheduler to the DAG, where it belongs.
  • The Scheduler is just a special type of DagRunJob. Users may want to implement their own "Schedulers" or their own scheduling logic, and by moving it to the DAG that becomes possible.

Description of New Workflow

DagRuns represent the state of a DAG at a certain point in time (perhaps they should be called DagInstances?). To run a DAG – or to manage the execution of a DAG – a DagRun must first be created. This can be done manually (simply by creating a DagRun object) or automatically, using methods like dag.schedule_dag(). Therefore, both scheduling new runs OR introducing ad-hoc runs can be done by any process at any time, simply by creating the appropriate object.

Just creating a DagRun is not enough to actually run the DAG (just as creating a TaskInstance is not the same as actually running a task). We need a Job for that. The DagRunJob is fairly simple in structure. It maintains a set of DagRuns that it is tasked with executing, and loops over that set until all the DagRuns either succeed or fail. New DagRuns can be passed to the job explicitly via DagRunJob.submit_dagruns() or by defining its DagRunJob.collect_dagruns() method, which is called during each loop. When the DagRunJob is executing a specific DagRun, it locks it. Other DagRunJobs will not try to execute locked DagRuns. This way, many DagRunJobs can run simultaneously in either a local or distributed setting, and can even be pointed at the same DagRuns, without worrying about collisions or interference.

The basic DagRunJob loop works like this:

1. refresh dags
2. collect new dagruns
3. process dagruns (including updating dagrun states for success/failure)
4. call executor/own heartbeat

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. 

Framework in progress

...

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

...