Airflow depends_on_past for whole DAG

user_7931431 0
Viewed 20120

Is there a way in airflow of using the depends_on_past for an entire DagRun, not just applied to a Task?

I have a daily DAG, and the Friday DagRun errored on the 4th task however the Saturday and Sunday DagRuns still ran as scheduled. Using depends_on_past = True would have paused the DagRun on the same 4th task, however the first 3 tasks would still have run.

I can see in the DagRun DB table there is a state column that contains failed for the Friday DagRun. What I want is a way configuring a DagRun to not start if the previous DagRun failed, not start and run until finding a Task that previously failed.

Does anyone know if this is possible?

apache-airflow airflow-scheduler airflow
4 Answers

This question is a bit old but it turns out as a first google search result and the highest rated answer is clearly misleading (and it has made me struggle a bit) so it definitely demands a proper answer. Although the second rated answer should work, there's a cleaner way to do this and I personally find using xcom ugly.

The Airflow has a special operator class designed for monitoring status of tasks from other dag runs or other dags as a whole. So what we need to do is to add a task preceding all the tasks in our dag, checking if the previous run has succeded.

from airflow.sensors.external_task_sensor import ExternalTaskSensor


previous_dag_run_sensor = ExternalTaskSensor(
    task_id = 'previous_dag_run_sensor',
    dag = our_dag,
    external_dag_id = our_dag.dag_id,
    execution_delta = our_dag.schedule_interval
)

previous_dag_run_sensor.set_downstream(vertices_of_indegree_zero_from_our_dag)
user_55255800

Here a solution that addresses Marc Lamberti's concern, namely, that 'wait_for_download' is not "recursive".

The solution entails "embedding" your original DAG in between two dummy tasks, a start_task and an end_task.

Such that:

  • The start_task precedes all your original initial tasks (ie, no other task in your DAG can start until start_task is completed).
  • A end_task follows all your original ending tasks (ie, all branches in your DAG converge to that dummy end_task).
  • start_task also directly precedes the end_task.

These conditions are provided by the following code:

start_task >> [_all_your_initial_tasks_here_]
[_all_your_ending_tasks_here] >> end_task
start_task >> end_task

Additionally, one needs to set that start_task has depends_on_past=True and wait_for_downstream=True

user_23813290
Related