从S3存储桶读取大量小型CSV文件的高效方法问询

1. 精简pandas.read_csv的参数配置 #

每次调用read_csv默认会执行类型推断、空行检查等冗余操作，针对小文件可关闭这些步骤减少开销：

• 指定low_memory=False：避免pandas多次重复推断数据类型
• 显式定义dtype参数：提前指定各列数据类型，省去类型检测时间（例如dtype={'user_id': 'int32', 'score': 'float32'}）
• 用usecols限定读取列：只加载业务需要的字段，减少内存占用与IO消耗
• 关闭skip_blank_lines：若确认文件无空行，设为False跳过空行判断

示例代码：

import pandas as pd
from glob import glob

file_paths = glob('./small_csvs/*.csv')
# 提前定义数据类型映射
dtype_config = {'user_id': 'int32', 'score': 'float32'}
# 列表推导式批量读取，减少循环开销
df_list = [pd.read_csv(f, dtype=dtype_config, low_memory=False, usecols=['user_id', 'score']) for f in file_paths]
final_df = pd.concat(df_list, ignore_index=True)

2. 并行化读取文件 #

单线程循环读取的效率受限于IO与CPU，可通过多进程/多线程并行处理：

• 使用concurrent.futures.ProcessPoolExecutor：利用多进程规避GIL限制，适合IO+CPU混合场景
• 进程数建议设为CPU核心数的1-2倍，避免过度上下文切换

示例代码：

import pandas as pd
from glob import glob
from concurrent.futures import ProcessPoolExecutor

def read_single_csv(file_path):
    dtype_config = {'user_id': 'int32', 'score': 'float32'}
    return pd.read_csv(file_path, dtype=dtype_config, low_memory=False)

file_paths = glob('./small_csvs/*.csv')
with ProcessPoolExecutor(max_workers=4) as executor:
    df_list = list(executor.map(read_single_csv, file_paths))
final_df = pd.concat(df_list, ignore_index=True)

3. 预合并小文件（长期优化方案） #

如果小文件是定期生成的，建议先合并为大文件再读取，彻底消除多次IO的开销：

• 用命令行工具快速合并（比Python代码效率更高）：

  # Linux/macOS：保留第一个文件的表头，合并其余文件内容
  head -n 1 ./small_csvs/first.csv > merged.csv
  tail -n +2 ./small_csvs/*.csv >> merged.csv
  
  # Windows PowerShell：同样处理重复表头
  Get-Content ./small_csvs/first.csv | Select-Object -First 1 | Out-File merged.csv -Encoding utf8
  Get-Content ./small_csvs/*.csv | Select-Object -Skip 1 | Out-File merged.csv -Encoding utf8 -Append
  

合并完成后直接读取单个大文件，速度会提升数倍。

4. 替换为更高效的IO库 #

针对大量小文件，一些轻量级库的读取开销远低于pandas：

• 使用polars库：专为大数据场景优化的DataFrame库，读取小文件的初始化开销更低
• 使用原生csv模块：读取后再转换为DataFrame，跳过pandas的封装层

示例（polars）：

import polars as pl
from glob import glob

file_paths = glob('./small_csvs/*.csv')
# polars支持直接传入文件路径列表批量读取
df = pl.read_csv(file_paths, dtype={'user_id': pl.Int32, 'score': pl.Float32})
# 如需转为pandas格式
final_df = df.to_pandas()

5. 优化存储介质访问 #

如果文件存储在网络介质（NAS、云存储）上，连接建立是核心瓶颈：

• 批量下载所有小文件到本地磁盘后再读取，避免多次网络连接开销
• 使用云存储SDK的批量操作接口（如S3的批量下载），减少连接建立次数

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