如何基于pandas DataFrame预估待写入CSV文件的大小

靠memory_usage()、getsizeof()拿到的是DataFrame在进程内存里的结构化存储大小，和序列化后存到磁盘的CSV文本大小没有固定换算比例：数值类型转成字符串长度不固定、文本字段遇到逗号/换行/引号会加转义符、不同编码下中文占的字节数也不一样，提前估算永远有误差，最靠谱的方案是写入时实时校验磁盘文件大小，到阈值自动切分，完全不需要提前换算。

这个方案准确率100%，还能避免全量数据加载到内存占资源，从数据库读数据阶段就可以分块拉取，不用一次性把整表读到内存里。
实现逻辑：

• 按固定小批次从数据库拉数据（每批几百到几千行）
• 往当前输出文件追加写入
• 每写完一批直接读取磁盘上的文件实际大小
• 大小接近阈值时自动新建下一个分片文件继续写

可直接运行的参考代码（适配Windows路径）：

import os
import pandas as pd
from sqlalchemy import create_engine

# 配置参数
DB_CONN = create_engine("mssql+pymssql://账号:密码@数据库地址/库名?charset=utf8")
SQL = "SELECT * FROM 你要导出的表"  # 支持带WHERE条件的自定义查询
OUTPUT_FOLDER = r"D:\导出文件存放目录"  # Windows路径前加r避免转义符冲突
MAX_SIZE = 32 * 1024 * 1024  # 单文件阈值32MB，单位为字节
SAFE_RATIO = 0.98  # 留2%冗余，避免最后一批写完超阈值
BATCH_ROWS = 1000  # 每次从数据库拉1000行处理，可根据机器内存调整

# 初始化第一个分片
part_num = 1
current_file = os.path.join(OUTPUT_FOLDER, f"export_part_{part_num}.csv")

for chunk in pd.read_sql(SQL, DB_CONN, chunksize=BATCH_ROWS):
    # 文件不存在就先写表头，存在就追加内容不重复写表头
    write_mode = "w" if not os.path.exists(current_file) else "a"
    header = True if write_mode == "w" else False
    chunk.to_csv(current_file, index=False, mode=write_mode, header=header, encoding="utf-8-sig")
    
    # 读取当前文件实际磁盘大小
    current_size = os.path.getsize(current_file)
    # 达到安全阈值就切换到下一个分片
    if current_size >= MAX_SIZE * SAFE_RATIO:
        part_num += 1
        current_file = os.path.join(OUTPUT_FOLDER, f"export_part_{part_num}.csv")

这个方案的优势：

• 大小判断完全基于磁盘实际写入的字节数，误差最多不超过单批次数据的CSV大小，基本不会出现超阈值的情况
• 内存占用极低，哪怕导出几千万行的表，内存里也只存当前批次的几千行数据，不会出现内存溢出
• os.path.getsize()是直接读取文件系统的元数据，没有额外IO开销，不会拖慢导出速度
• 自动适配所有数据类型，不管是数值、短文本还是超长备注字段，都不需要单独处理转义带来的大小变化

如果因为业务要求必须在写入前就把DataFrame拆分好，只能通过采样计算换算系数，不要直接用固定的5-10倍经验比例：

1. 从全量DataFrame里随机抽2%~5%的样本
2. 把样本写到临时CSV，算出样本CSV大小和样本内存占用的比值
3. 用这个比值估算全量CSV大小，再按比例拆分DataFrame
4. 所有分片写完后再做一次大小校验，对超标的分片做二次拆分

参考代码片段：

# 假设df是你已经加载好的全量DataFrame
import tempfile
# 抽3%的样本
sample_df = df.sample(frac=0.03)
# 写到临时文件计算大小
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False, encoding="utf-8-sig") as f:
    temp_path = f.name
sample_df.to_csv(temp_path, index=False)
sample_csv_size = os.path.getsize(temp_path)
sample_mem_usage = sample_df.memory_usage(deep=True).sum()
# 算出内存到CSV大小的换算系数
size_coeff = sample_csv_size / sample_mem_usage
estimated_total_csv_size = df.memory_usage(deep=True).sum() * size_coeff
# 后续按这个系数估算每个分片对应的行数即可
os.remove(temp_path)

这个方案缺陷很明显：如果数据里存在长度波动极大的字段（比如有的行备注是空，有的行是上万字的文本），采样得到的系数偏差会很大，拆分出来的文件大小很可能不符合预期，非必要不推荐使用。

• Windows下导出CSV建议用utf-8-sig编码，避免用Excel打开的时候出现中文乱码
• 单批次行数不要设太大，建议控制在500~5000行，批次越小，拆分后的文件大小越接近设定的阈值
• 不要把冗余比例设为100%，否则最后一批数据写入后很容易超出阈值，留2%~5%的冗余最稳妥

内容的提问来源于stack exchange，提问作者mugiwara-san