本文针对循环读取并合并大量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文件是瓶颈，可以考虑使用多线程并行读取，进一步提升效率。在实际应用中，需要根据具体情况选择合适的优化方案。
