从Cloud Storage读取Schema时BigQuery加载任务失败

故障原因 #

pandas的read_excel等IO方法内部集成了对GCS路径的支持，能自动处理gs://开头的URI，但BigQuery Python客户端的schema_from_json方法不直接支持GCS URI——它只能读取本地文件路径或者内存中的文件对象，直接传入GCS URI会触发FileNotFoundError。

正确读取方式 #

需要先通过Google Cloud Storage客户端将GCS上的Schema文件内容读取到内存，再传给schema_from_json。具体步骤：

• 拆分GCS URI，提取bucket名称和文件路径
• 使用Storage客户端获取对应Blob对象
• 将Blob内容读取为字符串，包装成内存文件对象
• 把内存文件对象传入schema_from_json

修正后的完整代码 #

from google.cloud import storage
from google.cloud import bigquery
import pandas as pd
import io
import datetime as dt  # 补上原代码缺失的datetime导入

bq_client = bigquery.Client()
storage_client = storage.Client()

def load_wwmd_raw(raw_table_id, schema_uri, source_uri):
    df = pd.read_excel(source_uri, engine='openpyxl')
    df['status'] = 'A'
    df['insert_timestamp'] = dt.datetime.now()
    df['update_timestamp'] = ''

    # 读取GCS中的Schema文件
    # 拆分GCS URI：去掉前缀后分割bucket和文件路径
    schema_uri_parts = schema_uri.replace('gs://', '').split('/', 1)
    bucket_name = schema_uri_parts[0]
    schema_blob_path = schema_uri_parts[1]
    
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(schema_blob_path)
    # 将Blob内容读取为字符串，包装成内存文件对象
    schema_file = io.StringIO(blob.download_as_text())
    
    # 从内存文件对象加载Schema
    schema = bq_client.schema_from_json(schema_file)

    # 设置Job配置
    job_config = bigquery.LoadJobConfig(
        schema=schema,
        write_disposition="WRITE_APPEND",
    )

    # 执行加载任务
    job = bq_client.load_table_from_dataframe(df, raw_table_id, job_config=job_config)
    job.result()

if __name__ == '__main__':
    project_id      = '<ProjectIDHere>'
    dataset_nm      = 'test_dataset'
    raw_table_nm    = 'test_raw'
    raw_table_id = f"{project_id}.{dataset_nm}.{raw_table_nm}"
    source_uri      = 'gs://<bucketName>/RAW_FILES/test_data.xlsx'
    schema_uri      = 'gs://<BucketName>/SCHEMA_FILES/RAW_SCHEMA.json'

    load_wwmd_raw(raw_table_id, schema_uri, source_uri)

额外注意点 #

• 确保运行代码的服务账号拥有GCS存储对象读取权限（roles/storage.objectViewer）和BigQuery表写入权限
• 原代码中source_uri未作为参数传入函数，改为参数传递可提升代码健壮性和复用性
• 若Schema文件体积过大，也可下载到本地临时文件后读取，但内存读取方式更高效

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