Airflow运行DAG报错：No module named 'airflow.providers'

报错原因 #

你遇到的No module named 'airflow.providers'错误，核心是Airflow版本不匹配：

• airflow.providers是Airflow 2.x版本引入的模块化结构，第三方服务的Operator统一放在该路径下；
• 若你的Cloud Composer环境用的是Airflow 1.x，这个路径不存在——1.x中Google Cloud相关Operator都在airflow.contrib.operators路径下。

解决方案 #

步骤1：确认Cloud Composer的Airflow版本 #

登录GCP控制台，进入你的Cloud Composer环境详情页查看Airflow版本，再对应以下操作：

情况1：环境为Airflow 1.x

修改导入语句，替换为1.x的路径，调整后的代码如下：

from datetime import datetime
from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryCreateEmptyTableOperator

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

# Create a DAG instance
with DAG(
    dag_id = "dag_bigquery",
    default_args= default_args,
    # schedule_interval='@daily',
    catchup = False
) as dag :
    
    # create empty table task
    create_table_task = BigQueryCreateEmptyTableOperator(
        task_id='create_table',
        project_id = "projectid",
        dataset_id = "datasetid",
        table_id= "himani-dummy-table",
        schema_fields=[
        {'name':'Brand_Id',"type": 'INT64'},
        {'name':'Brand_Code','type':'STRING'},
        {'name':'Brand_Name','type':'STRING'},
        {'name':'Entity_Id','type':'STRING'},
        {'name':'Entity_InsertedAt','type':'DATE'},
        {'name':'Entity_InsertedBy',"type": 'INT64'},
        {'name':'Entity_UpdatedAt','type':'DATE'},
        {'name':'Entity_UpdatedBy','type':'STRING'},
        {'name':'Entity_DivisionId','type':'STRING'},
        {'name':'Entity_OemId','type':'STRING'},
        ],
        bigquery_conn_id ='my-bigquery-connection'
    )

create_table_task

情况2：环境为Airflow 2.x

1. 确认apache-airflow-providers-google包已安装：Cloud Composer 2.x默认预装该包，若缺失，可通过环境的「添加PyPI包」功能添加（需指定与Airflow版本兼容的版本）；
2. 若导入语句正确仍报错，尝试重启Composer环境的Web服务器和调度器，确保包加载生效。

额外注意事项 #

• Cloud Composer 1.x已逐步停止维护，建议升级到2.x版本，获得更好的兼容性和功能支持；
• 确保Airflow连接中my-bigquery-connection已正确配置，拥有BigQuery访问权限。

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