本地Airflow使用BigQueryOperator报错：无法导入该类

错误信息 #

Traceback (most recent call last): File "c:\Users\hsaini\airflow-local\dags\bigQueryOperator.py", line 4, in <module> from airflow.providers.google.cloud.operators.bigquery import BigQueryOperator ImportError: cannot import name 'BigQueryOperator' from 'airflow.providers.google.cloud.operators.bigquery' (C:\Users\hsaini\AppData\Local\Programs\Python\Python311\Lib\site-packages\airflow\providers\google\cloud\operators\bigquery.py)

排查原因 #

1. Airflow与Google Provider版本不兼容
   Airflow主版本和apache-airflow-providers-google包版本必须严格匹配，否则会出现类找不到的情况。比如Python 3.11需要Airflow 2.5及以上版本，对应的Google Provider版本也需要适配该Airflow版本。

2. 包安装冲突或未生效
   重复执行pip install命令可能导致不同版本的包冲突，或者安装后未重启Airflow服务，新安装的包未被加载。

解决办法 #

1. 检查版本兼容性 #

执行以下命令查看当前安装的版本：

pip show apache-airflow apache-airflow-providers-google

确保两者版本匹配（比如Airflow 2.8.x对应apache-airflow-providers-google>=10.0.0，Airflow 2.7.x对应9.16.0版本）。

2. 重新安装兼容的Provider包 #

先卸载现有包：

pip uninstall -y apache-airflow-providers-google

然后安装与Airflow版本匹配的版本，示例：

# 适配Airflow 2.8.x
pip install apache-airflow-providers-google==10.0.0
# 适配Airflow 2.7.x
pip install apache-airflow-providers-google==9.16.0

3. 验证导入路径 #

Airflow 2.x的正确导入路径为：

from airflow.providers.google.cloud.operators.bigquery import BigQueryOperator

你的代码中该路径是正确的，无需修改；注释中的旧路径是Airflow 1.x的写法，可忽略。

4. 修正代码中的SQL错误 #

你的SQL语句存在语法问题，会导致任务执行失败：

• BigQuery中表名需用反引号`包裹，而非单引号
• 插入查询结果时无需使用VALUES，直接用INSERT ... SELECT

修正后的完整代码：

from airflow import DAG
from datetime import datetime, timedelta
from airflow.providers.google.cloud.operators.bigquery import BigQueryOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 4, 20),
}

with DAG(
    dag_id="dag_bigquery",
    default_args=default_args,
    catchup=False
) as dag:
    
    create_table_task = BigQueryOperator(
        task_id='create_table',
        sql="""
        CREATE TABLE `project_name.dataset_name.himani-dummy-table` 
        (Brand_Id INT, Brand_Code STRING, Brand_Name STRING, Entity_Id STRING)
        """,
        bigquery_conn_id='my-bigquery-connection',
    )

    load_data_task = BigQueryOperator(
        task_id='load_data',
        sql="""
        INSERT INTO `project_name.dataset_name.himani-dummy-table` 
        (Brand_Id, Brand_Code, Brand_Name, Entity_Id)
        SELECT Brand_Id, Brand_Code, Brand_Name, Entity_Id 
        FROM `project_name.dataset_name.table_name` LIMIT 10
        """,
        bigquery_conn_id='my-bigquery-connection',
    )

create_table_task >> load_data_task

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