Airflow Scheduler: the very basics

dag_params = {
  ...
  start_date: datetime(2018, 6, 6),
  end_date: None, # Explicitly makes the DAG open-ended, run forever
  ...
 }
 
 with DAG('Foo', default_args=dag_params, schedule_interval='0 2 * * *') as dag:
  ...
  # a bunch of tasks
  ...

The first run

def following_schedule(self, dttm):
  ...
  cron = croniter(self._schedule_interval, dttm)
  following = timezone.make_aware(cron.get_next(datetime), self.timezone)
  ...
 
 next_run_date = following_schedule(start_date) # start_date=datetime(2018, 6, 6). For our DAG this will yield datetime(2018, 6, 6, 2)

period_end = following_schedule(next_run_date) # period_end=datetime(2018, 6, 7, 2)

1. next_run_date <= timezone.utcnow() # next_run_date is execution_date
2. period_end <= timezone.utcnow()

DAGRun

DAGRun(
  ...
  execution_date=next_run_date, # NOTE THIS HERE!
  start_date=timezone.utcnow(),
  ...
 )

DAGRun(
  ...
  execution_date=datetime(2018, 6, 6, 2),
  start_date=datetime(2018, 6, 7, 2, 0, 1, 45), # the start_date will be slightly after 2018-06-07T14:00:00 because the scheduler periodically checks if a DAGRun should run
  ...
 )

Subsequent DAGRuns

next_dag_run = following_schedule(last_scheduled_run)
 

Rationale

Wrapping it up