Wiring top-level DAGs together

by

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.

More Related Contents:

  1. Airflow 1.9.0 is queuing but not launching tasks
  2. Problem with start date and scheduled date in Apache Airflow
  3. Airflow – How to pass xcom variable into Python function
  4. Airflow: how to delete a DAG?
  5. For Apache Airflow, How can I pass the parameters when manually trigger DAG via CLI?
  6. Airflow: changing the crontab time for a DAG in Airflow
  7. Why is it recommended against using a dynamic start_date in Airflow?
  8. Create and use Connections in Airflow operator at runtime [duplicate]
  9. Make custom Airflow macros expand other macros
  10. Apache airflow macro to get last dag run execution time
  11. How to get last two successful execution dates of Airflow job?
  12. execution_date in airflow: need to access as a variable
  13. Airflow BashOperator return exit code 0 even when task failed and return exit code 1
  14. Pip error even Microsoft Visual C++ 14.0 is installed
  15. How to submit Spark jobs to EMR cluster from Airflow?
  16. Proper way to create dynamic workflows in Airflow
  17. setting up s3 for logs in airflow
  18. Airflow: how to get pip packages installed via their docker-compose.yml?
  19. How to create a conditional task in Airflow
  20. How to run Airflow on Windows
  21. pass task instance variables as op_args
  22. How to prevent airflow from backfilling dag runs?
  23. How to control the parallelism or concurrency of an Airflow installation?
  24. Airflow “This DAG isnt available in the webserver DagBag object “
  25. Apache Airflow scheduler does not trigger DAG at schedule time
  26. Airflow s3 connection using UI

Leave a Comment