Python 3.12+Pandas处理Zip中超大CSV的优化方案咨询

核心思路 #

不需要将60GB的p.csv完整解压到磁盘，而是通过流式读取原zip包内p.csv的末尾内容获取目标行，再以流式写入的方式构建新zip文件，全程大部分数据在内存/流中传输，仅处理必要的末尾部分和新行，大幅节省磁盘IO开销。

具体实现步骤 #

1. 读取p.csv的倒数第二行 #

利用zipfile模块直接读取zip包内p.csv的末尾字节，避免加载整个大文件：

import zipfile

def get_last_second_line(zip_path, csv_name):
    # 读取末尾10KB内容（可根据单行长度调整大小）
    read_size = 1024 * 10
    with zipfile.ZipFile(zip_path, 'r') as zf:
        with zf.open(csv_name) as f:
            # 跳到文件末尾前read_size的位置
            f.seek(-read_size, 2)
            # 读取末尾内容并解码为文本
            tail = f.read().decode('utf-8')
            # 分割成行，处理可能的不完整行
            lines = tail.splitlines()
            # 取倒数第二行（若末尾有空行则取倒数第三行，需根据实际文件格式调整）
            return lines[-2] if len(lines) >= 2 else lines[-1] if lines else ''

last_second_line = get_last_second_line('t.zip', 'p.csv')

2. 基于目标行生成新行 #

用Pandas处理单行数据，生成需要追加的新行：

import pandas as pd
from io import StringIO

# 将单行转为DataFrame（无表头时需指定header=None）
df_last = pd.read_csv(StringIO(last_second_line), header=None)
# 执行自定义业务逻辑，示例：修改最后一列数值
df_new = df_last.copy()
df_new.iloc[0, -1] = df_new.iloc[0, -1] + 1
# 将新行转为字符串（不含表头、索引）
new_line = df_new.to_csv(header=False, index=False, line_terminator='')

3. 构建新的zip文件 #

遍历原zip的所有条目，直接复制小文件s.csv，对p.csv采用流式读写：

import shutil

with zipfile.ZipFile('t.zip', 'r') as src_zf, zipfile.ZipFile('t_new.zip', 'w', zipfile.ZIP_DEFLATED) as dest_zf:
    # 直接复制s.csv到新zip
    dest_zf.writestr('s.csv', src_zf.read('s.csv'))
    
    # 流式处理p.csv：复制原内容后追加新行
    with src_zf.open('p.csv') as src_f, dest_zf.open('p.csv', 'w') as dest_f:
        # 分块复制原文件内容（10MB块大小，可根据内存调整）
        shutil.copyfileobj(src_f, dest_f, length=1024*1024*10)
        # 追加新行（注意换行符需与原文件保持一致）
        dest_f.write(b'\n' + new_line.encode('utf-8'))

注意事项 #

• 若原p.csv末尾已有换行符，需调整追加时的换行符逻辑，避免出现空行；
• 分块大小length可根据内存情况灵活调整，越大传输效率越高但占用内存越多；
• 此方案仍会重新压缩p.csv，但全程在流中完成，无需写入本地磁盘，比原方案节省大量磁盘IO时间和空间；
• 若要完全避免重新压缩p.csv主体内容，需深入操作zip底层结构，复杂度极高，不适用于普通场景。

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