本文介绍针对小规模CSV文件（如6列×5行）批量合并的优化策略，重点规避list.append()累积DataFrame带来的内存与时间开销，推荐使用生成器式读取+单次pd.concat()、分治合并或流式写入等高性能方案。
本文介绍针对小规模csv文件（如6列×5行）批量合并的优化策略，重点规避`list.append()`累积dataframe带来的内存与时间开销，推荐使用生成器式读取+单次`pd.concat()`、分治合并或流式写入等高性能方案。
在处理大量小型CSV文件（例如每文件仅6列×5行，含1行表头）时，采用“逐个读取 → append() 到列表 → 最后pd.concat()”的模式虽直观，却存在显著性能缺陷：
- 每次
list.append()本身开销虽小，但当文件数达数百或上千时，Python列表动态扩容引发多次内存复制； - 更关键的是，
pd.DataFrame对象在Python层是重量级对象，频繁创建和引用会加剧GC压力与内存碎片； -
pd.concat()对长列表执行时，内部需遍历全部DataFrame并校验schema一致性，效率随列表长度非线性下降。
✅ 推荐优化方案（按优先级排序）：
1. 使用生成器 + 单次pd.concat()（最简高效）
避免中间列表，直接将read_csv调用惰性封装为生成器，交由pd.concat()统一处理：
import pandas as pd
import os
from pathlib import Path
def read_csv_generator(file_paths, **read_csv_kwargs):
for fp in file_paths:
yield pd.read_csv(fp, **read_csv_kwargs)
# 示例：每8个文件一组合并
all_files = []
for root, _, files in os.walk(base_dir):
for file in files:
if file.endswith('.csv'):
all_files.append(os.path.join(root, file))
# 分组处理（每8个一组）
for i in range(0, len(all_files), 8):
batch = all_files[i:i+8]
# 关键：不构建列表，直接传生成器
combined_df = pd.concat(
read_csv_generator(batch, sep='\t'),
ignore_index=True,
sort=False
)
combined_df.to_csv(f"batch_{i//8 + 1}.csv", index=False, sep='\t')✅ 优势：零中间列表、内存占用恒定（仅当前DataFrame）、代码简洁、充分利用
pd.concat的底层优化。
QuickBooks Mileage CSV下载使用 qbo-mileage CLI 及用户凭证，从 Airtable、Outlook 或 Google Calendar 记录生成 QuickBooks Online 里程 CSV 文件。
2. 分治式两两合并（适合超大文件集或内存受限场景）
当文件总数极大（如>10,000）且单次concat仍显吃力时，可模拟归并排序逻辑，递归两两合并临时文件：
def merge_two_csv(path1, path2, output_path, **kwargs):
df1 = pd.read_csv(path1, **kwargs)
df2 = pd.read_csv(path2, **kwargs)
pd.concat([df1, df2], ignore_index=True).to_csv(output_path, index=False, **kwargs)
def merge_files_in_batches(file_list, batch_size=8, temp_prefix="temp_"):
# 第一阶段：每batch_size个文件合并为一个临时文件
temp_files = []
for i in range(0, len(file_list), batch_size):
batch = file_list[i:i+batch_size]
temp_out = f"{temp_prefix}{i}.csv"
pd.concat(
(pd.read_csv(f, sep='\t') for f in batch),
ignore_index=True
).to_csv(temp_out, index=False, sep='\t')
temp_files.append(temp_out)
# 第二阶段：两两合并临时文件（可迭代进行，无需递归函数）
while len(temp_files) > 1:
new_temp_files = []
for i in range(0, len(temp_files), 2):
if i + 1 < len(temp_files):
merged = f"{temp_prefix}_merged_{i}.csv"
merge_two_csv(temp_files[i], temp_files[i+1], merged, sep='\t')
new_temp_files.append(merged)
else:
new_temp_files.append(temp_files[i])
temp_files = new_temp_files
return temp_files[0] # 最终合并结果路径3. 流式写入（零内存DataFrame，极致轻量）
若后续分析不强依赖Pandas（如仅需统计或导出），可跳过DataFrame，直接用csv模块拼接内容（跳过重复表头）：
import csv
def fast_concat_csv_files(file_list, output_path, sep='\t'):
with open(output_path, 'w', newline='') as out_f:
writer = csv.writer(out_f, delimiter=sep)
for i, fp in enumerate(file_list):
with open(fp, 'r') as f:
reader = csv.reader(f, delimiter=sep)
rows = list(reader)
if i == 0:
# 写入首文件的表头
writer.writerows(rows)
else:
# 跳过后续文件的表头（rows[0]），只写数据行
writer.writerows(rows[1:])
# 调用示例
fast_concat_csv_files(batch, "output.csv")⚠️ 注意事项：
- 所有方案中，确保各CSV文件列名、列顺序、数据类型严格一致，否则
pd.concat可能隐式转换或报错；建议在读取时显式指定dtype和usecols；- 若文件编码不统一，务必在
read_csv中添加encoding='utf-8'（或自动检测）；- 对于您描述的“每模型8个输出”的业务逻辑，建议将
batch_size=8与模型ID绑定（如按文件名前缀分组），避免跨模型混拼。
综上，摒弃list.append()累积DataFrame是性能跃升的关键第一步。对典型的小型CSV场景，方案1（生成器+单次concat）在简洁性、可维护性与性能间取得最佳平衡，应作为首选实践。
