Wiring top-level DAGs together

Question

I need to have several identical (differing only in arguments) top-level DAGs that can also be triggered together with following constraints / assumptions:

• Individual top-level DAGs will have schedule_interval=None as they will only need occasional manual triggering
• The series of DAGs, however, needs to run daily
• Order and number of DAGs in series is fixed (known ahead of writing code) and changes rarely (once in a few months)
• Irrespective of whether a DAG fails or succeeds, the chain of triggering must not break
• Currently they must be run together in series; in future they may require parallel triggering

---

So I created one file for each DAG in my dags directory and now I must wire them up for sequential execution. I have identified two ways this could be done:

1. SubDagOperator

   • Works without a glitch in my demo
   • Can lead to deadlocks but there are easy solutions; still there's a lot of haze around using them
   • SubDag's dag_id must be prefixed by it's parent's, that would force absurd IDs on top-level DAGs that are supposed to be functional independently too

2. TriggerDagRunOperator

   • Works in my demo but runs in parallel (not sequentially) as it doesn't wait for triggered DAG to finish before moving onto next one
   • ExternalTaskSensor might help overcome above limitation but it would make things very messy

---

My questions are

• How to overcome limitation of parent_id prefix in dag_id of SubDags?
• How to force TriggerDagRunOperators to await completion of DAG?
• Any alternate / better way to wire-up independent (top-level) DAGs together?
• Is there a workaround for my approach of creating separate files (for DAGs that differ only in input) for each top-level DAG?

---

I'm using puckel/docker-airflow with

• Airflow 1.9.0-4
• Python 3.6-slim
• CeleryExecutor with redis:3.2.7

---

EDIT-1

Clarifying @Viraj Parekh's queries

Can you give some more detail on what you mean by awaiting completion of the DAG before getting triggered?

When I trigger the import_parent_v1 DAG, all the 3 external DAGs that it is supposed to fire using TriggerDagRunOperator start running parallely even when I chain them sequentially. Actually the logs indicate that while they are fired one-after another, the execution moves onto next DAG (TriggerDagRunOperator) before the previous one has finished. NOTE: In this example, the top-level DAGs are named as importer_child_v1_db_X and their corresponding task_ids (for TriggerDagRunOperator) are named as importer_v1_db_X

Would it be possible to just have the TriggerDagRunOperator be the last task in a DAG?

I have to chain several similar (differing only in arguments) DAGs together in a workflow that triggers them one-by-one. So there isn't just one TriggerDagRunOperator that I could put at last, there are many (here 3, but would be upto 15 in production)

Answer 1

Taking hints from @Viraj Parekh's answer, I was able to make TriggerDagRunOperator work in the intended fashion. I'm hereby posting my (partial) answer; will update as and when things become clear.

---

How to overcome limitation of parent_id prefix in dag_id of SubDags?

As told @Viraj, there's no straight way of achieving this. Extending SubDagOperator to remove this check might work but I decided to steer clear of it

---

How to force TriggerDagRunOperators to await completion of DAG?

• Looking at the implementation, it becomes clear that the job of TriggerDagRunOperator is just to trigger external DAG; and that's about it. By default, it is not supposed to wait for completion of DAG. Therefore the behaviour I'm observing is understandable.

• ExternalTaskSensor is the obvious way out. However while learning basics of Airflow I was relying on manual triggering of DAGs (schedule_interval=None). In such case, ExternalTaskSensor makes it difficult to accurately specify execution_date for the external task (who's completion is being awaited), failing which the sensor gets stuck.

• So taking hint from implementation, I made minor adjustment to behaviour of ExternalTaskSensor by awaiting completion of all task_instances of concerned task having

  execution_date[external_task] >= execution_date[TriggerDagRunOperator] + execution_delta

  This achieves the desired result: external DAGs run one-after-other in sequence.

---

Is there a workaround for my approach of creating separate files (for DAGs that differ only in input) for each top-level DAG?

Again going by @Viraj this can be done by assigning DAGs to global scope using globals()[dag_id] = DAG(..)

---

EDIT-1

Maybe I was referring to incorrect resource (the link above is already dead), but ExternalTaskSensor already includes the params execution_delta & execution_date_fn to easily restrict execution_date(s) for the task being sensed.

Answer 2

• Can you give some more detail on what you mean by awaiting completion of the DAG before getting triggered? Would it be possible to just have the TriggerDagRunOperator be the last task in a DAG?

• For creating DAGs similar DAGs, you can dynamically generate the DAGs from one Python file. You could do something like this:

from airflow import DAG

from airflow.operators.python_operator import PythonOperator


def create_dag(dag_id,
               schedule,
               dag_number,
               default_args):

def hello_world_py(*args):
    print('Hello World')
    print('This is DAG: {}'.format(str(dag_number)))

dag = DAG(dag_id,
          schedule_interval=schedule,
          default_args=default_args)

with dag:
    t1 = PythonOperator(
        task_id='hello_world',
        python_callable=hello_world_py,
        dag_number=dag_number)

return dag


# build a dag for each number in range(10)
for n in range(1, 10):
dag_id = 'hello_world_{}'.format(str(n))

default_args = {'owner': 'airflow',
                'start_date': datetime(2018, 1, 1)
                }

schedule = '@daily'

dag_number = n

globals()[dag_id] = create_dag(dag_id,
                              schedule,
                              dag_number,
                              default_args)

You can read more about that approach here. If most of you are producing DAGs are pretty similar, you might want to consider storing the config in an Airflow Variableenter link description here

You probably won't be able to overcome the prefix limitations of the SubDag Operator - I'd suggest removing SubDags from your workflows entirely and just have them run as separate DAGs - it'll make it much easier to go back and re-run older DagRuns if you ever find yourself having to do that.

Answer 3

This worked for me when I was using schedule none.

trigger_dag = TriggerDagRunOperator(
    task_id=f'dag_id-trigger',
    trigger_dag_id='dag_id',
    python_callable=set_args,
    dag=dag,
)


def get_most_recent_dag_run(execution_date, **kwargs):
    return DagRun.find(dag_id='dag_id').pop().execution_date


sensor = ExternalTaskSensor(
    task_id=f'dag_id-sensor',
    external_dag_id='dag_id',
    execution_date_fn=get_most_recent_dag_run,
    dag=dag,
    poke_interval=5,
    external_task_id='last_task_id' # This is task need to be in your external dag
)

Answer 4

If you are looking for a way to wait for the triggered DAG completion, in Airflow 2.0 it became much easier than before.
There is a new version of the TriggerDagRunOperator allowing you to that. No need to use the ExternalTaskSensor anymore.
I've made a 10 mins tutorial about it https://youtu.be/8uKW0mPWmCk

Enjoy