Airflow ExternalTaskSensor持续轮询依赖DAG问题排查求助

Let me walk through the common issues that could be causing your DAG B to endlessly poll DAG A, and fix your configuration step by step:

1. You're Not Specifying the Correct Allowed States #

You set allowed_states=None, which is likely throwing off the sensor's logic. By default, ExternalTaskSensor waits for the external task to reach the ['success'] state—but when you explicitly set this to None, the sensor loses that clear target. Instead of stopping when DAG A's end task succeeds, it might be waiting for an undefined set of states, leading to infinite polling.

Fix: Explicitly set allowed_states=['success'] to tell the sensor exactly when to stop waiting.

2. Execution Date Mismatch #

Your execution_delta and execution_date_fn are both set to None, which means the sensor uses DAG B's current execution date to look for a matching run of DAG A. If:

• DAG A and DAG B have different schedules (e.g., DAG A runs daily at midnight, DAG B runs daily at 8 AM),
• Or DAG A's execution date doesn't align with DAG B's (like if DAG A runs with a delay),

the sensor won't find a successful end task for the matching execution date, so it keeps polling.

Fix: Use either execution_delta or execution_date_fn to align the execution dates. For example, if DAG A runs 2 hours before DAG B:

execution_delta=timedelta(hours=2)

Or if you need more dynamic logic, use execution_date_fn to calculate the correct execution date for DAG A.

3. Verify the External Task Exists & Reaches Success #

Double-check that:

• DAG A (datamart_OTT_CMS_v1) actually has a task with task_id="end",
• This end task consistently reaches the success state when DAG A completes (it's not being skipped, failed, or stuck in a running state).

Corrected Sensor Configuration #

Here's your updated code with the critical fixes:

from datetime import timedelta
from airflow.sensors.external_task import ExternalTaskSensor

TaskSensor = ExternalTaskSensor(
    task_id='TaskSensor',
    external_dag_id='datamart_OTT_CMS_v1',
    external_task_id="end",
    # Explicitly wait for success state
    allowed_states=['success'],
    # Adjust this delta to match your DAG schedule offset
    execution_delta=timedelta(hours=2),
    dag=dag
)

If you still see polling after these changes, check Airflow's logs for the sensor—they should tell you which execution date it's looking for and why it can't find a matching successful task.

内容的提问来源于stack exchange，提问作者K.he