Python中SpooledTemporaryFile适配DictWriter与BigQuery上传的问题

方案一：用TextIOWrapper封装二进制临时文件 #

核心思路是用二进制模式打开SpooledTemporaryFile，再通过io.TextIOWrapper将其转换为文本流供DictWriter使用，同时保留二进制底层流满足BigQuery的加载要求。

from csv import DictReader, DictWriter
from tempfile import SpooledTemporaryFile
from google.cloud.bigquery import Client, LoadJobConfig, SourceFormat
import io

csv_lines = ["A;2;3", "B;4;6", "C;8;12"]
fieldnames = ["Foo", "Bar", "Baz"]
csv_reader = DictReader(csv_lines, fieldnames=fieldnames, delimiter=";")

with SpooledTemporaryFile(mode="w+b") as tmp:
    # 将二进制流包装为文本流，适配DictWriter的文本写入需求
    text_stream = io.TextIOWrapper(tmp, encoding="utf-8", newline="")
    writer = DictWriter(text_stream, fieldnames=fieldnames)
    
    for row in csv_reader:
        writer.writerow({"Foo": row["Foo"], "Bar": row["Bar"], "Baz": row["Baz"]})
    
    # 刷新文本流缓冲区，确保数据写入底层二进制文件
    text_stream.flush()
    # 将文件指针移到开头，让BigQuery能读取完整内容
    tmp.seek(0)

    job = Client().load_table_from_file(
        tmp,
        "GCP_PROJECT_ID.BIGQUERY_DATASET_ID.BIGQUERY_TABLE_ID",
        job_config=LoadJobConfig(
            source_format=SourceFormat.CSV, skip_leading_rows=0, autodetect=True
        ),
    )
    job.result()

• io.TextIOWrapper自动处理文本到字节的编码转换，解决DictWriter与二进制流的兼容性问题
• 必须调用text_stream.flush()，避免文本数据滞留在缓冲区未写入底层文件
• tmp.seek(0)重置文件指针是关键，否则BigQuery会从当前指针位置开始读取，导致数据缺失

方案二：直接使用BytesIO内存缓冲区（更轻量） #

如果数据量不大，完全可以用io.BytesIO替代SpooledTemporaryFile，全程在内存中操作，彻底规避磁盘I/O：

from csv import DictReader, DictWriter
from google.cloud.bigquery import Client, LoadJobConfig, SourceFormat
import io

csv_lines = ["A;2;3", "B;4;6", "C;8;12"]
fieldnames = ["Foo", "Bar", "Baz"]
csv_reader = DictReader(csv_lines, fieldnames=fieldnames, delimiter=";")

# 用BytesIO作为纯内存缓冲区
buffer = io.BytesIO()
# 包装为文本流供DictWriter写入
text_stream = io.TextIOWrapper(buffer, encoding="utf-8", newline="")
writer = DictWriter(text_stream, fieldnames=fieldnames)

for row in csv_reader:
    writer.writerow({"Foo": row["Foo"], "Bar": row["Bar"], "Baz": row["Baz"]})

# 刷新缓冲区并重置指针到开头
text_stream.flush()
buffer.seek(0)

job = Client().load_table_from_file(
    buffer,
    "GCP_PROJECT_ID.BIGQUERY_DATASET_ID.BIGQUERY_TABLE_ID",
    job_config=LoadJobConfig(
        source_format=SourceFormat.CSV, skip_leading_rows=0, autodetect=True
    ),
)
job.result()

• BytesIO是纯内存对象，比SpooledTemporaryFile更高效，适合小数据场景
• 同样需要通过TextIOWrapper转换为文本流，写完后必须重置指针

通用注意事项 #

• 无论采用哪种方案，写完数据后都必须重置文件/缓冲区指针到开头（seek(0)），否则BigQuery无法读取完整数据
• TextIOWrapper的newline参数设为空字符串，可确保CSV换行符处理符合标准，避免跨平台格式问题

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