GCP BlobWriter写入CSV格式失效问题求助

你当前的代码直接将字典序列化为JSON字符串写入，导致生成的文件是JSON对象拼接的格式，而非标准CSV。要解决这个问题，需要使用Python的csv模块将字典转换为CSV格式后再流式写入。

方案一：使用csv.DictWriter（推荐，适配字典结构） #

这种方式能自动匹配字典键与CSV列，适配结构不一致的字典场景：

import csv
import io

blob = defaults.bucket.blob(filename)
csv_writer = BlobWriter(blob, content_type="text/csv")

events = result.get('events', [])
# 收集所有字典的键作为CSV表头，兼容结构不一致的情况
all_headers = set()
for data in events:
    _source = data.get('_source', {})
    all_headers.update(_source.keys())
headers = sorted(all_headers)  # 排序确保表头顺序固定

# 写入CSV表头
header_io = io.StringIO()
dict_writer = csv.DictWriter(header_io, fieldnames=headers)
dict_writer.writeheader()
csv_writer.write(header_io.getvalue().encode('utf-8'))

# 逐行流式写入字典数据
for data in events:
    _source = data.get('_source', {})
    row_io = io.StringIO()
    dict_writer = csv.DictWriter(row_io, fieldnames=headers)
    # 自动按表头匹配字典键，缺失字段填空字符串
    dict_writer.writerow(_source)
    csv_writer.write(row_io.getvalue().encode('utf-8'))

方案二：使用csv.writer（手动控制字段顺序） #

如果所有字典结构完全一致，可直接取第一个字典的键作为表头：

import csv
import io

blob = defaults.bucket.blob(filename)
csv_writer = BlobWriter(blob, content_type="text/csv")

events = result.get('events', [])
headers = None

for data in events:
    _source = data.get('_source', {})
    # 首次获取表头，取第一个非空字典的键
    if not headers and _source:
        headers = list(_source.keys())
        # 写入表头
        header_io = io.StringIO()
        csv_header_writer = csv.writer(header_io)
        csv_header_writer.writerow(headers)
        csv_writer.write(header_io.getvalue().encode('utf-8'))
    
    # 写入当前行数据
    if headers:
        row_io = io.StringIO()
        csv_row_writer = csv.writer(row_io)
        # 按表头顺序提取对应值，缺失字段填空字符串
        row_values = [_source.get(key, '') for key in headers]
        csv_row_writer.writerow(row_values)
        csv_writer.write(row_io.getvalue().encode('utf-8'))

关键改动说明 #

• 替换json.dumps为csv模块的写入逻辑，将字典映射为标准CSV行数据
• 使用io.StringIO作为中间缓冲区，生成CSV格式字符串后编码为字节写入BlobWriter
• 处理表头生成：支持单一首例键或多字典合并键两种场景，避免数据缺失
• 自动填充缺失字段为空字符串，保证CSV格式完整性

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