ExternalTaskSensor for cross-DAG dependencies in Apache Airflow - Mini Project: Build & Apply

ExternalTaskSensor for cross-DAG dependencies

📖 Scenario: You are managing workflows in Apache Airflow. You have two DAGs: dag_a and dag_b. dag_b should wait for a specific task in dag_a to complete before starting its own tasks. This is a common real-world need when workflows depend on each other.

🎯 Goal: Build an Airflow DAG dag_b that uses ExternalTaskSensor to wait for the task task_in_dag_a in dag_a to finish before running its own task.

📋 What You'll Learn

Create a DAG called dag_b with default arguments

Use ExternalTaskSensor to wait for task_in_dag_a in dag_a

Add a simple PythonOperator task called task_in_dag_b that runs after the sensor

Print a message in task_in_dag_b to confirm it runs after the sensor

💡 Why This Matters

🌍 Real World

In real projects, workflows often depend on other workflows. ExternalTaskSensor helps coordinate these dependencies safely.

💼 Career

Understanding cross-DAG dependencies is essential for Airflow users and DevOps engineers managing complex data pipelines.

1

Create the DAG dag_b with default arguments

Write code to import DAG from airflow and create a DAG called dag_b with start_date set to datetime(2024, 1, 1) and schedule_interval set to @daily. Use default_args with owner set to 'airflow'.

Apache Airflow

# Import DAG and datetime
# Create default_args dictionary
# Create DAG object named dag_b
# Your code here

Hint

Use from airflow import DAG and datetime from datetime module. Define default_args as a dictionary with 'owner': 'airflow'. Then create dag_b with these arguments.

2

Add an ExternalTaskSensor to wait for task_in_dag_a in dag_a

Import ExternalTaskSensor from airflow.sensors.external_task. Create a task called wait_for_task_in_dag_a using ExternalTaskSensor that waits for the task task_in_dag_a in DAG dag_a. Set dag parameter to dag_b.

Apache Airflow

from airflow.sensors.external_task import ExternalTaskSensor

# Create ExternalTaskSensor task named wait_for_task_in_dag_a
# Your code here

Hint

Use ExternalTaskSensor with task_id='wait_for_task_in_dag_a', external_dag_id='dag_a', and external_task_id='task_in_dag_a'. Pass dag=dag_b to assign the task to the DAG.

3

Add a PythonOperator task called task_in_dag_b

Import PythonOperator from airflow.operators.python. Define a Python function called print_message that prints 'Task in dag_b is running'. Create a PythonOperator task named task_in_dag_b that calls print_message and belongs to dag_b. Set task_in_dag_b to run after wait_for_task_in_dag_a.

Apache Airflow

from airflow.operators.python import PythonOperator

# Define function print_message
# Create PythonOperator task task_in_dag_b that calls print_message
# Set task_in_dag_b to run after wait_for_task_in_dag_a
# Your code here

Hint

Define a function print_message that prints the message. Use PythonOperator with task_id='task_in_dag_b' and python_callable=print_message. Use wait_for_task_in_dag_a >> task_in_dag_b to set the order.

4

Print confirmation message when task_in_dag_b runs

Run the DAG dag_b and observe the output of task_in_dag_b. It should print Task in dag_b is running after wait_for_task_in_dag_a completes.

Apache Airflow

# Run the DAG and check the logs of task_in_dag_b
# Your code here

Hint

Trigger dag_b in Airflow UI or CLI. Check the logs of task_in_dag_b to see the printed message.

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