使用Airflow将多个GCS文件导入BigQuery时遇无效表ID错误求助

GoogleCloudStorageToBigQueryOperator的destination_project_dataset_table参数不支持通配符，它要求明确指定单个BigQuery表的完整标识符（格式为项目ID.数据集ID.表名）。你写的project_name.dataset.*会被BigQuery判定为无效表ID，因为表名不能包含*字符。

要实现“GCS文件名对应BigQuery表名”的多文件加载，分两种场景处理：

场景1：Airflow 2.2及以上版本（推荐用动态任务映射） #

利用Airflow的动态任务映射特性，自动为每个CSV文件生成对应的加载任务：

from airflow import DAG
from airflow.providers.google.cloud.operators.gcs import GoogleCloudStorageListOperator
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GoogleCloudStorageToBigQueryOperator
from airflow.utils.dates import days_ago

BUCKET_NAME = 'my_bucket'
PROJECT_ID = 'project_name'
DATASET_ID = 'dataset'

with DAG(
    'dag_sensor',
    default_args=dict(start_date=days_ago(1)),
    schedule_interval='@daily',
    catchup=False,
) as dag:
    # 1. 获取GCS桶内所有目标CSV文件
    list_csv_files = GoogleCloudStorageListOperator(
        task_id='list_gcs_csv_files',
        bucket=BUCKET_NAME,
        delimiter='.csv',  # 过滤出CSV文件
        google_cloud_storage_conn_id='google_cloud_default',
    )

    # 2. 动态生成每个文件的BigQuery加载任务
    load_file_to_bq = GoogleCloudStorageToBigQueryOperator.partial(
        task_id='write_to_bq',
        bucket=BUCKET_NAME,
        source_format='CSV',
        skip_leading_rows=1,
        autodetect=True,
        write_disposition='WRITE_TRUNCATE',
        create_disposition='CREATE_IF_NEEDED',
        google_cloud_storage_conn_id='google_cloud_default',
        bigquery_conn_id='google_cloud_default',
        allow_jagged_rows=True,
    ).expand(
        # 为每个文件指定源路径
        source_objects=list_csv_files.output,
        # 将文件名转换为BigQuery表名（去掉.csv后缀）
        destination_project_dataset_table=list_csv_files.output.map(
            lambda filename: f"{PROJECT_ID}.{DATASET_ID}.{filename.replace('.csv', '')}"
        )
    )

    list_csv_files >> load_file_to_bq

场景2：Airflow版本低于2.2（手动循环创建任务） #

如果你的Airflow版本不支持动态映射，可手动遍历文件名列表创建任务：

from airflow import DAG
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GoogleCloudStorageToBigQueryOperator
from airflow.utils.dates import days_ago

BUCKET_NAME = 'my_bucket'
PROJECT_ID = 'project_name'
DATASET_ID = 'dataset'
# 指定要加载的CSV文件列表
TARGET_FILES = ['order_comments.csv', 'order_users.csv']

with DAG(
    'dag_sensor',
    default_args=dict(start_date=days_ago(1)),
    schedule_interval='@daily',
    catchup=False,
) as dag:
    for file_name in TARGET_FILES:
        # 从文件名提取表名（去掉.csv后缀）
        table_name = file_name.replace('.csv', '')
        GoogleCloudStorageToBigQueryOperator(
            task_id=f'write_to_bq_{table_name}',
            bucket=BUCKET_NAME,
            source_objects=[file_name],
            source_format='CSV',
            destination_project_dataset_table=f'{PROJECT_ID}.{DATASET_ID}.{table_name}',
            skip_leading_rows=1,
            autodetect=True,
            write_disposition='WRITE_TRUNCATE',
            create_disposition='CREATE_IF_NEEDED',
            google_cloud_storage_conn_id='google_cloud_default',
            bigquery_conn_id='google_cloud_default',
            allow_jagged_rows=True,
        )

• 如果CSV文件在GCS的子目录下，source_objects需要填写完整路径（比如['subdir/order_comments.csv']）
• autodetect=True会自动检测表结构，但如果字段类型有异常，建议手动指定schema_fields参数
• write_disposition='WRITE_TRUNCATE'会覆盖目标表现有数据，若需要追加数据可改为WRITE_APPEND

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