Airflow如何捕获BigQueryOperator执行SQL的错误并汇总展示

问题场景 #

需要构建Airflow DAG遍历SQL文件列表，执行对应BigQuery SQL，但部分SQL关联的BQ表不存在。要求：

• 不将错误打印到日志
• 不标记任务失败
• 收集错误信息，最后由专门任务统一展示

原方案用try-except包裹BigQueryOperator无效：错误仍会输出到日志、标记任务失败，最终错误字典为空。核心原因是Airflow的DAG解析阶段仅创建Operator对象，SQL执行逻辑在Worker运行阶段，解析阶段的try-except无法捕获执行阶段的错误。

核心原因 #

Airflow的DAG运行分为两个阶段：

1. 解析阶段：调度器加载DAG文件，创建所有Operator对象，不执行SQL
2. 执行阶段：Worker节点运行任务，才会触发Operator的核心执行逻辑（比如BigQueryOperator执行SQL）

原代码的try-except仅在解析阶段生效，只能捕获Operator初始化时的异常（如参数错误），完全触及不到SQL执行阶段的表不存在错误。

解决方案 #

将BigQuery的执行逻辑封装到PythonOperator的自定义函数中，在函数内部执行SQL并捕获错误，通过Airflow的XCom机制跨任务传递错误信息，最后统一收集展示。

完整代码实现 #

import airflow
from airflow import models
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime, timedelta
from google.cloud import bigquery, storage
from google.api_core.exceptions import NotFound

# 从Airflow变量获取配置
BQ_PROJECT = models.Variable.get('bq_datahub_project_id').strip()
BQ_DATASET_PREFIX = models.Variable.get('datasetPrefix').strip()
AIRFLOW_BUCKET = models.Variable.get('airflow_bucket').strip()
CODE = models.Variable.get('gcs_code').strip()
COMPOSER = '-'.join(models.Variable.get('airflow_bucket').strip().split('-')[2:-2])

# DAG默认参数
default_dag_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
}

def execute_sql_and_catch_errors(**context):
    """执行SQL并捕获错误，将错误信息推送到XCom"""
    sql_file = context['templates_dict']['sql_file']
    
    # 从GCS读取SQL文件内容（若为本地文件可替换为open()读取）
    bucket_name, blob_path = sql_file.replace('gs://', '').split('/', 1)
    storage_client = storage.Client()
    blob = storage_client.bucket(bucket_name).blob(blob_path)
    sql_content = blob.download_as_text()
    
    # 替换SQL中的宏参数
    sql_content = sql_content.format(
        datahubProject=BQ_PROJECT,
        datasetPrefix=BQ_DATASET_PREFIX,
        COMPOSER=COMPOSER
    )
    
    client = bigquery.Client(project=BQ_PROJECT, location='europe-west3')
    error_info = None
    
    try:
        query_job = client.query(sql_content)
        query_job.result()  # 等待SQL执行完成
    except NotFound as e:
        # 捕获表不存在的目标错误
        error_info = {'file': sql_file, 'error': f"表不存在: {str(e)}"}
    except Exception as e:
        # 捕获其他执行错误
        error_info = {'file': sql_file, 'error': f"执行失败: {str(e)}"}
    
    if error_info:
        # 将错误信息推送到XCom，供后续任务读取
        context['ti'].xcom_push(key='sql_exec_error', value=error_info)

def collect_and_print_errors(**context):
    """从XCom收集所有错误并打印"""
    errors = []
    ti = context['ti']
    
    # 遍历所有SQL任务，拉取XCom中的错误信息
    for task_id in context['dag'].task_ids:
        if task_id.startswith('sql_'):
            try:
                error_data = ti.xcom_pull(task_ids=task_id, key='sql_exec_error')
                if error_data:
                    errors.append(error_data)
            except Exception:
                pass
    
    if errors:
        print("=== SQL文件执行错误汇总 ===")
        for err in errors:
            print(f"文件路径: {err['file']}")
            print(f"错误详情: {err['error']}")
            print("------------------------")
    else:
        print("所有SQL文件执行成功，无错误。")

with models.DAG(dag_id='run-sql-files',
                schedule_interval='0 */3 * * *',
                user_defined_macros={"COMPOSER": COMPOSER},
                default_args=default_dag_args,
                concurrency=2,
                max_active_runs=1,
                catchup=False) as dag:
    start_task = DummyOperator(task_id='Start')
    end_task = DummyOperator(task_id='End', trigger_rule='all_done')
    
    # 替换为你的SQL文件列表（后续可从GCP存储动态获取）
    sql_files = ['gs://your-bucket/sql/file1.sql', 'gs://your-bucket/sql/file2.sql']
    
    # 为每个SQL文件创建执行任务
    for idx, sql_file in enumerate(sql_files):
        sql_exec_task = PythonOperator(
            task_id=f'sql_{idx}',
            python_callable=execute_sql_and_catch_errors,
            templates_dict={'sql_file': sql_file},
            provide_context=True,
            dag=dag
        )
        start_task >> sql_exec_task >> end_task
    
    # 错误汇总展示任务
    error_report_task = PythonOperator(
        task_id='print_errors',
        python_callable=collect_and_print_errors,
        provide_context=True,
        dag=dag
    )
    end_task >> error_report_task

关键说明 #

1. PythonOperator封装执行逻辑：将BigQuery的SQL执行放到Python函数中，确保错误捕获发生在任务执行阶段
2. XCom传递错误信息：Airflow任务运行在独立进程/容器中，无法共享全局变量，XCom是官方跨任务数据传递方案
3. 精准捕获错误类型：通过NotFound异常类精准捕获表不存在的错误，其他错误可按需调整处理逻辑
4. 任务状态控制：函数内部捕获所有异常，任务始终标记为成功，符合"不标记任务失败"的要求
5. 动态SQL读取：示例中从GCS读取SQL文件，若为本地文件可替换为open(sql_file).read()

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