如何用Airflow ExternalTaskSensor实现BigQuery表依赖调度及排查超时问题

要让表A的DAG等待8个依赖DAG全部执行完成后再加载数据，可按以下方式实现：

• 为每个依赖DAG创建独立的ExternalTaskSensor任务，分别监听对应DAG的目标任务完成状态。
• 将所有传感器任务设为加载BigQuery表A任务的前置任务，也可使用TaskGroup统一管理这些传感器，让DAG结构更清晰。
• 若依赖DAG触发时间不一致，需通过execution_date_fn参数自定义时间匹配逻辑，确保传感器能识别到依赖DAG的最新成功运行实例。

示例代码（Airflow 2.x）：

from airflow import DAG
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='load_table_a',
    default_args=default_args,
    description='Load data to BigQuery Table A after all dependent DAGs complete',
    schedule_interval=None,
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    # 用TaskGroup统一管理所有依赖传感器
    with TaskGroup('wait_for_dependencies') as wait_for_dependencies:
        # 依赖DAG 1的传感器
        wait_dag1 = ExternalTaskSensor(
            task_id='wait_for_dag1',
            external_dag_id='dependent_dag_1',  # 替换为实际依赖DAG的ID
            external_task_id='target_task_in_dag1',  # 替换为依赖DAG中需等待的任务ID
            allowed_states=['success'],
            mode='reschedule',  # 用reschedule模式节省Worker资源
            poke_interval=300,  # 每5分钟轮询一次
            timeout=3600,  # 超时时间设为1小时
        )

        # 重复上述代码，创建wait_dag2到wait_dag8，对应剩余7个依赖DAG

    # 加载数据到BigQuery表A的任务
    load_table_a = BigQueryInsertJobOperator(
        task_id='load_data_to_table_a',
        configuration={
            'query': {
                'query': 'INSERT INTO `project.dataset.table_a` SELECT * FROM ...',  # 替换为实际加载逻辑
                'useLegacySql': False,
            }
        },
        gcp_conn_id='google_cloud_default',
    )

    # 设置依赖关系：所有传感器完成后才执行加载任务
    wait_for_dependencies >> load_table_a

针对你提供的代码，超时失败可从以下几点逐一排查：

1. 语法错误：代码中poke_interval = 60后缺少逗号，会直接导致Python语法报错，修正后代码如下：

operator = ExternalTaskSensor(
    task_id='wait_for_dag',
    external_task_id="Task_dependent",
    external_dag_id='dependent_table',  # 注意：dag_id建议避免空格，需与实际依赖DAG的ID完全一致
    allowed_states=['success'],  # 若仅需等待成功，建议移除'failed'
    mode='poke',
    poke_interval=60,
    timeout=180
)

2. execution_date不匹配：默认情况下，ExternalTaskSensor会用当前DAG的execution_date匹配依赖DAG的execution_date。若手动触发表A的DAG，其execution_date为触发时间，而依赖DAG的execution_date是自身定时触发时间，两者不匹配会导致传感器找不到对应任务实例，持续轮询直到超时。

   • 解决方法：自定义execution_date_fn获取依赖DAG的最新成功实例时间：

     from airflow.models import DagRun
     
     def get_latest_success_execution_date(dt):
         dag_runs = DagRun.find(dag_id='dependent_table', state='success')
         if dag_runs:
             return max(dr.execution_date for dr in dag_runs)
         return dt
     
     operator = ExternalTaskSensor(
         # 其他参数保持不变
         execution_date_fn=get_latest_success_execution_date,
     )
     

3. 参数值错误：

   • 确认external_dag_id是否与依赖DAG的实际ID完全一致（注意大小写、空格、拼写）。
   • 确认external_task_id是否为依赖DAG中真实存在且会执行完成的任务ID，拼写错误会导致传感器无法找到目标任务。

4. 超时时间过短：timeout=180仅3分钟，若依赖DAG运行时间较长，或轮询间隔（60秒）导致还未等到任务完成就超时，建议调大超时时间（如timeout=3600即1小时）。

5. allowed_states设置不合理：若设置allowed_states=['success', 'failed']，意味着依赖任务失败也会触发传感器完成，但实际你需要的是依赖任务成功完成，建议仅保留['success']，避免依赖失败后误触发加载任务。

内容的提问来源于stack exchange，提问作者delusional_reality