本文针对循环读取并合并大量CSV文件至Pandas DataFrame时效率低下的问题，提供了基于数据收集和多线程的优化方案。通过将数据暂存至字典后一次性合并，以及利用`ThreadPoolExecutor`实现并行读取，显著提升数据处理速度，并附带代码示例和注意事项。
在数据分析和处理过程中，经常会遇到需要读取大量CSV文件并将它们合并成一个大的DataFrame的情况。如果文件数量巨大，传统的循环读取并逐个合并的方法效率会非常低下，尤其是在每次循环中使用pd.concat时，由于其会创建新的DataFrame对象，导致内存占用和计算时间显著增加。本文将介绍两种优化方案：一是避免在循环中使用pd.concat，而是先将数据收集起来，最后一次性合并；二是利用多线程并行读取文件，进一步提升效率。
优化方案一：数据收集与一次性合并
问题的关键在于避免在循环中频繁调用pd.concat。可以将每次读取的数据先存储在一个Python字典中，然后在循环结束后，使用pd.concat一次性将字典中的所有数据合并成一个DataFrame。
以下是优化后的代码示例：
import pathlibimport 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 ThreadPoolExecutorimport pathlibimport 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文件是瓶颈，可以考虑使用多线程并行读取，进一步提升效率。在实际应用中，需要根据具体情况选择合适的优化方案。
以上就是高效处理大量CSV文件：Pandas DataFrame分块与多线程优化的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1381082.html
微信扫一扫 
支付宝扫一扫 
