
本文针对循环读取并合并大量CSV文件至Pandas DataFrame时效率低下的问题,提供了基于数据收集和多线程的优化方案。通过将数据暂存至字典后一次性合并,以及利用`ThreadPoolExecutor`实现并行读取,显著提升数据处理速度,并附带代码示例和注意事项。
在数据分析和处理过程中,经常会遇到需要读取大量CSV文件并将它们合并成一个大的DataFrame的情况。如果文件数量巨大,传统的循环读取并逐个合并的方法效率会非常低下,尤其是在每次循环中使用pd.concat时,由于其会创建新的DataFrame对象,导致内存占用和计算时间显著增加。本文将介绍两种优化方案:一是避免在循环中使用pd.concat,而是先将数据收集起来,最后一次性合并;二是利用多线程并行读取文件,进一步提升效率。
优化方案一:数据收集与一次性合并
问题的关键在于避免在循环中频繁调用pd.concat。可以将每次读取的数据先存储在一个Python字典中,然后在循环结束后,使用pd.concat一次性将字典中的所有数据合并成一个DataFrame。
以下是优化后的代码示例:
import pathlib
import pandas as pd
# 假设 root_path 是CSV文件所在的根目录
root_path = pathlib.Path('root')
# 创建一个字典来存储读取的数据
data = {}
# 循环读取CSV文件
for count, (_, row) in enumerate(df.iterrows(), 1):
folder_name = row['File ID'].strip()
file_name = row['File Name'].strip()
file_path = root_path / folder_name / file_name
folder_file_id = f'{folder_name}_{file_name}'
# 读取CSV文件,并设置列名
file_data = pd.read_csv(file_path, header=None, sep='\t',
names=['Case', folder_file_id],
memory_map=True, low_memory=False)
# 将读取的数据存储到字典中,键为folder_file_id
data[folder_file_id] = file_data.set_index('Case').squeeze()
print(count)
# 使用pd.concat一次性合并所有数据
merged_data = (pd.concat(data, names=['folder_file_id'])
.unstack('Case').reset_index())代码解释:
- 使用pathlib.Path代替os.path,使代码更具可读性和跨平台性。
- 使用enumerate代替手动计数器,使代码更简洁。
- 使用字典data来存储每次读取的DataFrame。
- 在循环结束后,使用pd.concat一次性合并字典中的所有DataFrame。
- memory_map=True, low_memory=False 参数可以帮助处理大型CSV文件。
- .set_index('Case').squeeze() 将'Case'列设置为索引,并将DataFrame压缩成Series,以减少内存占用。
- .unstack('Case').reset_index() 将'Case'列取消堆叠,并将索引重置为默认的整数索引。
注意事项:
- 确保所有CSV文件具有相同的列结构,以便合并。
- 根据实际情况调整memory_map和low_memory参数。
- 如果CSV文件非常大,可以考虑使用dask等库进行分块读取和并行处理。
优化方案二:多线程并行读取
如果读取CSV文件是瓶颈,可以考虑使用多线程并行读取。Python的concurrent.futures模块提供了ThreadPoolExecutor,可以方便地实现多线程。
以下是使用多线程的代码示例:
from concurrent.futures import ThreadPoolExecutor
import pathlib
import pandas as pd
# 假设 root_path 是CSV文件所在的根目录
root_path = pathlib.Path('root')
def read_csv(args):
count, row = args # expand arguments
folder_name = row['File ID'].strip()
file_name = row['File Name'].strip()
file_path = root_path / folder_name / file_name
folder_file_id = f'{folder_name}_{file_name}'
file_data = pd.read_csv(file_path, header=None, sep='\t',
names=['Case', folder_file_id],
memory_map=True, low_memory=False)
print(count)
return folder_file_id, file_data.set_index('Case').squeeze()
with ThreadPoolExecutor(max_workers=2) as executor:
batch = enumerate(df[['File ID', 'File Name']].to_dict('records'), 1)
data = executor.map(read_csv, batch)
merged_data = (pd.concat(dict(data), names=['folder_file_id'])
.unstack('Case').reset_index())代码解释:
- 定义一个read_csv函数,用于读取单个CSV文件。
- 使用ThreadPoolExecutor创建线程池,max_workers参数指定线程数量。
- 使用executor.map将read_csv函数应用到df的每一行,实现并行读取。
- df[['File ID', 'File Name']].to_dict('records') 将DataFrame转换为字典列表,方便传递给read_csv函数。
- enumerate 函数为每个文件添加一个计数器,用于打印进度。
注意事项:
- max_workers参数需要根据CPU核心数和实际情况进行调整。
- 多线程适用于I/O密集型任务,如果CPU计算是瓶颈,多线程可能无法带来显著的性能提升。
- 注意线程安全问题,避免多个线程同时访问共享资源。
总结
通过避免在循环中使用pd.concat,而是先将数据收集起来,最后一次性合并,可以显著提升处理大量CSV文件的效率。如果读取CSV文件是瓶颈,可以考虑使用多线程并行读取,进一步提升效率。在实际应用中,需要根据具体情况选择合适的优化方案。










