Airflow starts two DAG runs when turned on for the first time

Question

When I boot up the Airflow webserver and scheduler for the first time on Oct 25th at around 17:23, and turn on my DAG, I can see that it kicks off two runs for Oct 23rd and Oct 24th:

RUN 1 -> 10-23T17:23
RUN 2 -> 10-24T17:23

Here's my DAG configuration:

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': '2019-01-01',
    'retries': 0,
}
dag = DAG(
    'my_script',
    default_args=default_args,
    schedule_interval=datetime.timedelta(days=1),
    catchup=False,
)

Since it's past the start_date + schedule_interval and I have set catchup=False, I would expect it to kick off a single DAG run immediately, however I would not expect it to run two.

Why are two DAG runs being executed?
How can I prevent this behaviour?

Emma Emma · Accepted Answer · 2019-10-25T22:30:49

I am not sure but this is my best guess -

In short answer, could be it is how airflow is built and workaround would be to modify your start_date to be yesterday.

TL;DR

I agree that kicks off 1 dag for 10-24 when you turned on would sound more natural.

However, according to your dag runs, RUN 1 is 10-23. This suggests to me that initializing of the first run is not correct and I have looked into the scheduler code.

And I have a doubt on this line.

https://github.com/apache/airflow/blob/68b8ec5f415795e4fa4ff7df35a3e75c712a7bad/airflow/jobs/scheduler_job.py#L603

This is inside a function that create a dag run and setting the start date of the run.

# The logic is that we move start_date up until
# one period before, so that timezone.utcnow() is AFTER
# the period end, and the job can be created...
now = timezone.utcnow()

# This returns current time + schedule_interval. In your example, this will be tomorrow.
next_start = dag.following_schedule(now)

# This returns current time - schedule_interval. In your example, this will be yesterday.
last_start = dag.previous_schedule(now)

# tomorrow <= today should return False 
if next_start <= now:
    new_start = last_start
else:
    # and this will return last_start - schedule_interval which means 2 days ago.  
    # wondering if this is intended to be dag.previous_schedule(next_start)???
    new_start = dag.previous_schedule(last_start)

Airflow starts two DAG runs when turned on for the first time

1 Answers