如何将BQ导出的自定义排序纯文本文件写入GCS云存储

API 探路者

2026-9-29

解决方案

原代码核心错误点

临时文件操作逻辑错误：将fh.name赋值为fh.write()的返回值（写入的字节数），覆盖了原临时文件路径，后续无法定位到生成的文件
临时文件生命周期错误：在with tempfile.NamedTemporaryFile()代码块结束后，临时文件会被自动删除，出块后再访问文件路径必然失败
GCS上传逻辑错误：已经通过bucket.blob('test.csv')生成了Blob对象，后续重复调用bucket.blob()传入Blob对象属于参数错误，且upload_from_file的参数要求为文件对象/可读路径，原代码参数传递不符合要求
数据导出格式不符合要求：使用to_string生成的内容是空格对齐的表格，而非要求的|分隔的纯文本格式

修改后可运行代码

import pandas as pd
import tempfile
from google.colab import auth
from google.cloud import storage

# 账号鉴权
auth.authenticate_user()

# 此处job为你已按规则排序后的BQ查询结果
df = pd.DataFrame(data=job)

# 初始化GCS客户端
sc = storage.Client(project='temp-project')
bucket = sc.get_bucket('my-bucket')
destination_blob = bucket.blob('test.csv')

# 方案1：直接内存上传，无需本地临时文件（性能更高，推荐）
df_content = df.to_csv(sep='|', header=False, index=False)
destination_blob.upload_from_string(df_content, content_type='text/plain')

# 方案2：如果需要保留本地临时文件逻辑，可替换上面2行为下方代码
# import os
# with tempfile.NamedTemporaryFile(mode='w', delete=False, encoding='utf-8') as fh:
#     df.to_csv(fh, sep='|', header=False, index=False)
#     temp_file_path = fh.name
# destination_blob.upload_from_filename(temp_file_path)
# # 手动清理临时文件
# os.unlink(temp_file_path)