如何流式压缩大体积CSV文件？Python方案优先

核心思路 #

由于CSV是顺序存储文件，直接删除中间行效率极低且易损坏文件，采用**「处理一块→保留未处理部分→替换原CSV」**的方式，确保每一步只占用最少空间：

• 分块读取CSV，每次处理固定行数
• 将处理后的块追加写入Parquet/Feather（压缩率远高于CSV）
• 把原CSV中未处理的剩余内容写入临时文件，替换原CSV（相当于删除已处理行）
• 循环直到所有数据处理完成

依赖安装 #

先安装必要的Python库：

pip install pandas pyarrow chardet

• pandas：分块读取CSV
• pyarrow：处理Parquet/Feather的高效引擎
• chardet：自动检测CSV文件编码（避免乱码）

完整代码 #

import os
import shutil
import chardet
import pandas as pd

def convert_csv_to_compressed(input_csv, output_path, chunksize=1_000_000, output_format='parquet', compression='snappy'):
    # 自动检测CSV编码
    with open(input_csv, 'rb') as f:
        result = chardet.detect(f.read(10000))
        encoding = result['encoding'] or 'utf-8'
    
    # 初始化输出文件标记
    first_write = True
    
    while True:
        # 分块读取CSV
        chunk_iter = pd.read_csv(input_csv, chunksize=chunksize, encoding=encoding)
        try:
            chunk = next(chunk_iter)
        except StopIteration:
            # 所有数据处理完成，删除剩余空CSV
            os.remove(input_csv)
            break
        
        # 写入压缩格式文件
        if output_format == 'parquet':
            chunk.to_parquet(
                output_path,
                engine='pyarrow',
                compression=compression,
                mode='append' if not first_write else 'write'
            )
        elif output_format == 'feather':
            chunk.to_feather(
                output_path,
                engine='pyarrow',
                compression=compression,
                mode='append' if not first_write else 'write'
            )
        first_write = False
        
        # 保留未处理的CSV内容：将剩余迭代器内容写入临时文件
        temp_csv = input_csv + '.tmp'
        with open(temp_csv, 'w', encoding=encoding, newline='') as temp_f:
            # 剩余块已无表头，直接写入数据行
            for remaining_chunk in chunk_iter:
                remaining_chunk.to_csv(temp_f, index=False, header=False)
        
        # 替换原CSV为临时文件
        os.remove(input_csv)
        shutil.move(temp_csv, input_csv)

# 使用示例
if __name__ == '__main__':
    INPUT_CSV = 'large_data.csv'
    OUTPUT_FILE = 'compressed_data.parquet'
    CHUNKSIZE = 1_000_000  # 内存不足时可改为500_000或更小
    OUTPUT_FORMAT = 'parquet'  # 可选'feather'
    COMPRESSION = 'snappy'  # Parquet可选'gzip'，Feather可选'zstd'
    
    convert_csv_to_compressed(INPUT_CSV, OUTPUT_FILE, CHUNKSIZE, OUTPUT_FORMAT, COMPRESSION)

关键注意事项 #

• 分块大小调整：如果运行时内存不足，减小chunksize（比如改为500,000或200,000）
• 压缩选择：
  • snappy：速度最快，压缩率适中，适合快速处理场景
  • gzip/zstd：压缩率更高，处理速度稍慢，适合极致省空间的场景
• 断点续处理：如果处理中断，原CSV已保留未处理部分，重新运行代码即可从断点继续
• 备份建议：若有额外存储空间，建议先备份原CSV（避免处理出错导致数据丢失）

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