合并JSON转CSV时，如何按1GB上限拆分写入多文件

这个需求我在处理TB级数据导出的时候经常碰到，核心就是实时监控当前写入文件的大小，一旦达到设定的阈值（比如1GB）就自动切换到新文件，同时要保证每个新文件的CSV表头只写一次。下面用Python给你实现一个可复用的方案，完全兼容你现在的逐行写入逻辑：

核心思路拆解 #

• 提前定义分片阈值（1GB = 102410241024 字节，如果你想按十进制的1GB（1000^3）算也可以，调整数值就行）
• 封装一个分片写入器类，负责：
  • 管理当前打开的文件句柄
  • 检查文件大小是否超过阈值
  • 自动创建新文件并写入表头
  • 处理逐行写入的逻辑
• 原来的JSON转CSV的逻辑不变，只是把直接写入文件换成调用这个写入器的方法

完整代码实现 #

import os
import csv

class SplitCSVWriter:
    def __init__(self, base_filename, header, max_size_bytes=1024*1024*1024):
        self.base_filename = base_filename  # 基础文件名，比如"output"
        self.header = header  # CSV表头列表
        self.max_size = max_size_bytes  # 分片阈值，默认1GB
        self.current_file_index = 1
        self.current_writer = None
        self._open_new_file()  # 初始化打开第一个文件

    def _open_new_file(self):
        # 关闭之前的文件（如果存在）
        if self.current_writer:
            self.current_writer.file.close()
        
        # 生成新文件名，比如output_1.csv, output_2.csv
        new_filename = f"{self.base_filename}_{self.current_file_index}.csv"
        # 打开新文件，用newline=''避免Windows下的空行问题
        new_file = open(new_filename, 'w', newline='', encoding='utf-8')
        self.current_writer = csv.writer(new_file)
        # 写入表头
        self.current_writer.writerow(self.header)
        print(f"开始写入新文件: {new_filename}")

    def write_row(self, row_data):
        # 检查当前文件大小是否超过阈值
        if os.path.getsize(self.current_writer.file.name) >= self.max_size:
            self.current_file_index += 1
            self._open_new_file()
        
        # 写入数据行
        self.current_writer.writerow(row_data)

    def close(self):
        # 最后关闭当前文件
        if self.current_writer:
            self.current_writer.file.close()

# ------------------- 你的业务逻辑部分 -------------------
if __name__ == "__main__":
    # 示例表头，替换成你实际的CSV表头
    csv_header = ["id", "name", "value", "created_at"]
    # 初始化分片写入器，设置基础文件名和阈值
    split_writer = SplitCSVWriter("output", csv_header)

    try:
        # 这里替换成你读取JSON文件、转换数据的逻辑
        # 模拟逐行生成数据（实际是从JSON转换来的）
        for i in range(10000000):
            # 模拟转换后的CSV行数据
            row = [i, f"item_{i}", i*2, "2024-05-20"]
            split_writer.write_row(row)
    finally:
        # 确保最后关闭文件
        split_writer.close()

关键细节说明 #

1. 文件大小检查：用os.path.getsize直接读取文件实际大小，比自己累计写入的字节数更准确（避免编码、换行符等因素导致的计算误差）
2. 文件命名规则：自动生成output_1.csv、output_2.csv这样的序列文件名，方便后续批量处理
3. 资源安全：在finally块中调用close()，确保即使程序抛出异常，文件也能被正确关闭
4. 表头处理：每个新文件创建时自动写入表头，保证每个分片CSV都是完整可独立打开的

适配你的现有代码 #

把你原来的csv.writer写入逻辑，替换成split_writer.write_row(row_data)即可。比如原来你可能是这样写的：

with open("output.csv", "w", newline='') as f:
    writer = csv.writer(f)
    writer.writerow(header)
    for json_data in json_files:
        row = convert_json_to_row(json_data)  # 你的转换函数
        writer.writerow(row)

现在改成：

split_writer = SplitCSVWriter("output", header)
try:
    for json_data in json_files:
        row = convert_json_to_row(json_data)
        split_writer.write_row(row)
finally:
    split_writer.close()

这样就完美解决了大数据量下的内存问题，因为全程都是逐行写入，不会把所有数据加载到内存里，同时自动分片避免单个文件过大。

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