Running dags with different frequency | Airflow

Question

I have been evaluating airflow .I have this use case where I have a workflow that runs every hour to get hourly aggregates of the data. and another that runs every day to get daily aggregates of the same. Is it possible to create a combined workflow where the daily aggregate will run only if all the hourly aggregates have succeed in past day? I have seen that you can create sub dags but can the two dags run at a different frequency ? If yes How?

For sure! Look this post: http://stackoverflow.com/questions/38022323/how-to-set-dependencies-between-dags-in-airflow I think its very similiar. — p.magalhaes, Jul 03 '16 at 12:27

score 4 · Answer 1 · answered Aug 15 '16 at 22:16

Not sure how you want this to work but while there isn't a straightforward way of doing this, there are a few ways you could use the extensive suite of airflow operators to build such a dag.

Example you could make the hourly dags depend_on_past and then use a python branch operator to make the day aggregation task/dag be run/triggered at the end of the hourly dag for the last run of the day. Check out the PythonBranchOperator and the TriggerDagRunOperator.

You could also create your own sensor for the daily aggregator to make sure that all hourly dags for that day have succeeded. Check out ExternalTaskSensor for reference.

score 0 · Answer 2 · answered Nov 18 '19 at 17:50

It might be ugly, but using the PythonOperator there is a pretty straight forward way of doing it "behind the scenes":

dag = DAG('hourly_daily_update_v0',
          schedule_interval='@hourly')

hourly_update = PythonOperator(task_id='update_hourly_v0',
                               python_callable=update_hourly,
                               provide_context=True,
                               dag=dag)

daily_update = PythonOperator(task_id='update_daily_v0',
                               python_callable=update_daily,
                               provide_context=True,
                               dag=dag)

So you call both hourly and daily the Airflow way. However in the update_daily() call you can make a check for the hour:

def update_daily(**context):
    if context['execution_date'].hour == 0: # hour 0
        # Do all the things!
    else:
        # Do none of the things!

Airflow will successfully run update_daily() 24 times a day, but in reality it will only do the work once at hour 0. You can extend this however you please. Only problem is the slight step outside of Airflow's assumed pattern, which will cause some disinformation between hours 1 and 24.

Running dags with different frequency | Airflow

2 Answers2