Python如何实现总大小12GB的100个CSV文件更快加载到内存

1. 替换原生csv模块为更高效的解析库 #

原生Python的csv模块是纯Python实现的逐行解析逻辑，性能较低，可替换为底层用编译型语言实现的高性能解析库：

• pandas：底层用C实现CSV解析逻辑，速度是原生csv的3~5倍，支持指定仅加载需要的列，大幅降低IO和内存开销，适配代码示例：

import pandas as pd
import os
import time

mylist = []
start = time.time()
for csv_file in sorted(os.listdir('myfolder')):
    # 过滤非csv文件，避免读取Mac系统隐藏文件如.DS_Store
    if not csv_file.endswith('.csv'):
        continue
    # usecols指定只加载第2列，header参数根据你的CSV是否有表头调整
    df = pd.read_csv(os.path.join('myfolder', csv_file), usecols=[1], header=None)
    mylist.extend(df.iloc[:, 0].tolist())
print("elapsed {}".format(time.time()-start))

• polars：Rust实现的高性能DataFrame库，针对多核CPU优化，解析CSV速度比pandas快2~4倍，适配代码示例：

import polars as pl
import os
import time

mylist = []
start = time.time()
for csv_file in sorted(os.listdir('myfolder')):
    if not csv_file.endswith('.csv'):
        continue
    df = pl.read_csv(os.path.join('myfolder', csv_file), columns=[1], has_header=False)
    mylist.extend(df.to_series().to_list())
print("elapsed {}".format(time.time()-start))

2. 预序列化数据避免重复加载 #

由于你需要多次运行测试场景，无需每次都重新解析CSV，可将第一次加载完成的数据序列化存储为本地二进制文件，后续测试直接读取二进制文件即可，速度可提升数十倍：

• 最简单的方案用Python内置pickle：

# 第一次运行时存储加载好的列表数据
import pickle
with open('preloaded_data.pkl', 'wb') as f:
    pickle.dump(mylist, f)

# 后续所有测试场景直接读二进制文件，无需再解析CSV
with open('preloaded_data.pkl', 'rb') as f:
    mylist = pickle.load(f)

• 更大数据量可选用pyarrow、msgpack等更高效的序列化方案，文件体积更小，读写速度比pickle高30%以上。

3. 并行加载压榨多核性能 #

M1芯片具备多核计算能力，内存足够的前提下可以用多进程并行读取多个CSV文件，进一步压缩加载时间，示例：

from concurrent.futures import ProcessPoolExecutor
import polars as pl
import os
import time

def load_single_csv(csv_file):
    if not csv_file.endswith('.csv'):
        return []
    df = pl.read_csv(os.path.join('myfolder', csv_file), columns=[1], has_header=False)
    return df.to_series().to_list()

start = time.time()
csv_files = sorted(os.listdir('myfolder'))
with ProcessPoolExecutor() as executor:
    results = executor.map(load_single_csv, csv_files)
mylist = []
for res in results:
    mylist.extend(res)
print("elapsed {}".format(time.time()-start))

4. 原有代码的基础优化点 #

• 路径拼接使用os.path.join('myfolder', csv_file)替代直接字符串拼接，避免路径格式错误带来的异常开销
• 读取文件时明确指定encoding参数，避免自动检测编码的额外性能损耗
• 提前过滤非CSV文件，避免读取无效文件浪费时间

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