ExternalTaskSensor for cross-DAG dependencies in Apache Airflow - Commands & Configuration

Introduction

Sometimes you want one workflow to wait for another workflow to finish before starting. ExternalTaskSensor helps you do this by pausing a task until a task in a different workflow completes.

When you have two separate workflows and one needs to start only after the other finishes successfully.

When you want to coordinate data processing steps that happen in different DAGs.

When you want to avoid running a task before its required data is ready from another workflow.

When you want to build modular workflows that depend on each other without merging them into one big DAG.

Config File - cross_dag_sensor.py

cross_dag_sensor.py

from airflow import DAG
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.operators.dummy import DummyOperator
from datetime import datetime, timedelta

with DAG(
    dag_id='dependent_dag',
    start_date=datetime(2024, 6, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:

    wait_for_task = ExternalTaskSensor(
        task_id='wait_for_task_in_other_dag',
        external_dag_id='source_dag',
        external_task_id='final_task',
        allowed_states=['success'],
        failed_states=['failed', 'skipped'],
        mode='reschedule',
        poke_interval=30,
        timeout=600
    )

    proceed = DummyOperator(task_id='proceed_after_wait')

    wait_for_task >> proceed

This DAG named dependent_dag waits for a task called final_task in another DAG named source_dag to complete successfully before continuing.

ExternalTaskSensor pauses the workflow until the external task finishes with success.

mode='reschedule' lets Airflow free up worker slots while waiting.

poke_interval=30 means it checks every 30 seconds.

timeout=600 means it stops waiting after 10 minutes if the task never finishes.

Commands

List all available DAGs to confirm both source_dag and dependent_dag are present.

Terminal

airflow dags list

Expected Output

source_dag

dependent_dag

Start the source DAG to run its tasks, including the final_task that dependent_dag waits for.

Terminal

airflow dags trigger source_dag

Expected Output

Created <DagRun source_dag @ 2024-06-01T00:00:00+00:00: manual__2024-06-01T00:00:00+00:00, externally triggered: True>

List tasks in dependent_dag to verify the ExternalTaskSensor task and the next task.

Terminal

airflow tasks list dependent_dag

Expected Output

wait_for_task_in_other_dag

proceed_after_wait

Test the ExternalTaskSensor task manually for the given date to see if it detects the external task completion.

Terminal

airflow tasks test dependent_dag wait_for_task_in_other_dag 2024-06-01

Expected Output

[2024-06-01 00:00:00,000] {external_task.py:100} INFO - Poking for task final_task in DAG source_dag

[2024-06-01 00:00:30,000] {external_task.py:100} INFO - External task final_task in DAG source_dag succeeded

[2024-06-01 00:00:30,001] {taskinstance.py:1234} INFO - Task succeeded

Key Concept

If you remember nothing else from this pattern, remember: ExternalTaskSensor pauses a task until a specific task in another DAG finishes successfully.

Common Mistakes

Using ExternalTaskSensor with wrong external_dag_id or external_task_id.

The sensor will never find the task to wait for and will timeout or fail.

Double-check the exact DAG ID and task ID names in the other DAG before configuring the sensor.

Not setting mode='reschedule' for ExternalTaskSensor.

The sensor will occupy a worker slot the whole time it waits, reducing resource efficiency.

Always use mode='reschedule' to free up workers while waiting.

Not handling failed_states in ExternalTaskSensor.

If the external task fails or is skipped, the sensor might wait forever or behave unexpectedly.

Set failed_states=['failed', 'skipped'] to make the sensor fail fast if the external task does not succeed.

Summary

Use ExternalTaskSensor in one DAG to wait for a task in another DAG to finish.

Configure external_dag_id and external_task_id to point to the correct external workflow and task.

Set mode='reschedule' and handle allowed_states and failed_states for efficient and safe waiting.

Practice

1. What is the main purpose of the ExternalTaskSensor in Airflow?

easy

A. To wait for a task in another DAG to complete before proceeding

B. To trigger a DAG run manually

C. To schedule DAGs at specific intervals

D. To monitor system resource usage

Solution

1. Step 1: Understand the role of ExternalTaskSensor

   The ExternalTaskSensor is designed to pause a task until a specified task in a different DAG finishes successfully.

2. Step 2: Compare with other options

   Triggering DAGs, scheduling, and monitoring resources are handled by other components, not ExternalTaskSensor.

3. Final Answer:

   To wait for a task in another DAG to complete before proceeding -> Option A

4. Quick Check:

   ExternalTaskSensor waits for cross-DAG task completion [OK]

Hint: ExternalTaskSensor waits for tasks in other DAGs [OK]

Common Mistakes:

• Confusing ExternalTaskSensor with DAG trigger
• Thinking it schedules DAGs
• Assuming it monitors system resources

2. Which of the following is the correct way to instantiate an ExternalTaskSensor to wait for task 'task_a' in DAG 'dag_1'?

easy

A. ExternalTaskSensor(task_id='wait_task', dag='dag_1', task='task_a')

B. ExternalTaskSensor(task_id='wait_task', dag_id='dag_1', task_id='task_a')

C. ExternalTaskSensor(task_id='wait_task', external_dag='dag_1', external_task='task_a')

D. ExternalTaskSensor(task_id='wait_task', external_dag_id='dag_1', external_task_id='task_a')

Solution

1. Step 1: Identify correct parameter names

   The ExternalTaskSensor requires external_dag_id and external_task_id to specify the DAG and task it waits for.

2. Step 2: Match parameters with options

   Only ExternalTaskSensor(task_id='wait_task', external_dag_id='dag_1', external_task_id='task_a') uses the correct parameter names and syntax.

3. Final Answer:

   ExternalTaskSensor(task_id='wait_task', external_dag_id='dag_1', external_task_id='task_a') -> Option D

4. Quick Check:

   Use external_dag_id and external_task_id parameters [OK]

Hint: Use external_dag_id and external_task_id parameters [OK]

Common Mistakes:

• Using wrong parameter names like dag_id or external_dag
• Confusing task_id with external_task_id
• Omitting required parameters

3. Given this snippet in DAG 'dag_2':

wait_for_task = ExternalTaskSensor(
    task_id='wait_for_task_a',
    external_dag_id='dag_1',
    external_task_id='task_a',
    mode='reschedule',
    poke_interval=30
)

wait_for_task.execute(context={})

What happens if task_a in dag_1 has not completed yet?

medium

A. The sensor waits and retries every 30 seconds without blocking a worker slot

B. The sensor immediately fails with an error

C. The sensor completes successfully without waiting

D. The sensor blocks a worker slot and retries every 30 seconds

Solution

1. Step 1: Understand mode='reschedule'

   In 'reschedule' mode, the sensor frees the worker slot between pokes, retrying after the poke_interval.

2. Step 2: Analyze behavior when task not complete

   The sensor will keep checking every 30 seconds without blocking a worker slot until the external task finishes.

3. Final Answer:

   The sensor waits and retries every 30 seconds without blocking a worker slot -> Option A

4. Quick Check:

   mode='reschedule' frees worker slot during wait [OK]

Hint: mode='reschedule' frees worker slot during wait [OK]

Common Mistakes:

• Confusing 'reschedule' with 'poke' mode
• Assuming immediate failure if task incomplete
• Thinking sensor completes without waiting

4. You wrote this ExternalTaskSensor:

wait_task = ExternalTaskSensor(
    task_id='wait_task',
    external_dag_id='dag_1',
    external_task_id='task_a',
    timeout=300
)

But the sensor never finishes even after task_a completes. What is the likely cause?

medium

A. The timeout is too high

B. The sensor's execution_date does not match the external task's execution date

C. The task_id is incorrect

D. The sensor is missing poke_interval

Solution

1. Step 1: Understand execution_date matching

   ExternalTaskSensor waits for the external task with the same execution_date by default.

2. Step 2: Identify mismatch issue

   If the sensor's execution_date differs from the external task's, it will wait forever even if the external task completed.

3. Final Answer:

   The sensor's execution_date does not match the external task's execution date -> Option B

4. Quick Check:

   Execution dates must match for ExternalTaskSensor to succeed [OK]

Hint: Check execution_date alignment between DAGs [OK]

Common Mistakes:

• Assuming timeout controls waiting for task completion
• Confusing task_id with external_task_id
• Thinking poke_interval absence causes hang

5. You want to create a cross-DAG dependency where task_b in dag_2 waits for task_a in dag_1 to succeed, but only for the same execution date. Which configuration ensures this behavior and avoids deadlocks?

hard

A. Use ExternalTaskSensor with external_dag_id='dag_1', external_task_id='task_a', and execution_date_fn=lambda x: x - timedelta(days=1)

B. Use ExternalTaskSensor with external_dag_id='dag_1', external_task_id='task_a', and execution_date_fn=lambda x: x

C. Use ExternalTaskSensor with external_dag_id='dag_1', external_task_id='task_a', and default execution_date matching

D. Use ExternalTaskSensor with external_dag_id='dag_1', external_task_id='task_a', and mode='poke' without execution_date control

Solution

1. Step 1: Understand default execution_date behavior

   By default, ExternalTaskSensor waits for the external task with the same execution_date, ensuring correct cross-DAG sync.

2. Step 2: Avoid execution_date shifts

   Using execution_date_fn to shift dates can cause deadlocks or missed dependencies.

3. Step 3: Consider mode

   mode='poke' works but does not affect execution_date matching; default is fine here.

4. Final Answer:

   Use ExternalTaskSensor with external_dag_id='dag_1', external_task_id='task_a', and default execution_date matching -> Option C

5. Quick Check:

   Default execution_date matching avoids deadlocks [OK]

Hint: Default execution_date matching avoids deadlocks [OK]

Common Mistakes:

• Shifting execution_date causing missed dependencies
• Ignoring execution_date alignment
• Misusing mode without fixing execution_date