How to trigger Airflow DAG independent runs with an ExternalTaskSensor

user_9535682 0
Viewed 655

I built two DAGs(dag_a, dag_b) and create an ExternalTaskSensor in dag_b that pokes on dag_a. And there are two use cases for these DAGs:

  1. schedule dag_a and dag_b at the same time and use the dependency to process dag_a first and then dag_b
  2. manually trigger dag_b independently without caring about dag_a.

With an ExternalTaskSensor, use case 1 works pretty well. But use case 2 won't work. And the dag_b will stop at ExternalTaskSensor for dag_a and poke forever. Is there a way to skip ExternalTaskSensor with some conditions and run dag_b independently?

airflow
2 Answers

Since Airflow 2.0.0, I think there is no need to use ExternalTaskSensor. Using TriggerDagRunOperator, you could create and schedule a DAG that acts as a controller, having two tasks responsible for triggering DAG_A and DAG_B.

With the wait_for_completion param you could achieve your use case number one without affecting the possibility to trigger DAG_B independently. You could also specify the desire poke_interval and allowed or failed states.

Docs:

  • wait_for_completion (bool) -- Whether or not wait for dag run completion. (default: False)
  • poke_interval (int) -- Poke interval to check dag run status when wait_for_completion=True. (default: 60)
  • allowed_states (list) -- list of allowed states, default is ['success']
  • failed_states (list) -- list of failed or dis-allowed states, default is None

Here is a link to the example DAG provide with Airflow.

user_105692200

I was able to accomplish this by using a execution_date_fn to check whether the DAG run was manually triggered, and raising an AirflowSkipException if so. The sensor task will be skipped when the DAG is manually triggered.

Tested on Airflow 2.1.

from datetime import datetime

from airflow import DAG
from airflow.exceptions import AirflowSkipException
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import get_current_context
from airflow.sensors.external_task import ExternalTaskSensor


def get_execution_date(execution_date):
    context = get_current_context()
    if context["dag_run"].external_trigger:
        raise AirflowSkipException
    
    return execution_date


default_dag_args = {
    "schedule_interval": "* * * * *",
    "start_date": datetime(2022, 1, 1),
    "catchup": False,
}

with DAG(dag_id="dag_a", **default_dag_args) as dag_a:
    dummy_task = DummyOperator(task_id="dummy_task")

with DAG(dag_id="dag_b", **default_dag_args) as dag_b:
    await_dag_a = ExternalTaskSensor(
        task_id="await_dag_a",
        external_dag_id="dag_a.v1",
        execution_date_fn=get_execution_date,
    )
    
    dummy_task = DummyOperator(
        task_id="dummy_task",
        # enable task to trigger even if parent task is skipped
        trigger_rule="none_failed",
    )

    await_dag_a >> dummy_task
user_112088920
Related