Airflow: re execute the jobs of a DAG for the past n days on a daily basis

Question

I have scheduled the execution of a DAG to run daily. It works perfectly for one day.

However each day I would like to re-execute not only for the current day {{ ds }} but also for the previous n days (let's say n = 7).

For example, in the next execution scheduled to run on "2018-01-30" I would like Airflow not only to run the DAG using as execution date "2018-01-30", but also to re-run the DAGs for all the previous days from "2018-01-23" to "2018-01-30".

Is there an easy way to "invalidate" the previous execution so that a backfill is run automatically?

Answer 1

You can generate dynamically tasks in a loop and pass the offset to your operator.

Here is an example with the Python one.

import airflow
from airflow.operators.python_operator import PythonOperator
from airflow.models import DAG

from datetime import timedelta


args = {
    'owner': 'airflow',
    'start_date': airflow.utils.dates.days_ago(2),
    'schedule_interval': '0 10 * * *'
}

def check_trigger(execution_date, day_offset, **kwargs):
    target_date = execution_date - timedelta(days=day_offset)
    # use target_date

for day_offset in xrange(1, 8):
    PythonOperator(
        task_id='task_offset_' + i,
        python_callable=check_trigger,
        provide_context=True,
        dag=dag,
        op_kwargs={'day_offset' : day_offset}
    )