17370845950

本文针对循环读取并合并大量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())

代码解释：

1. 使用pathlib.Path代替os.path，使代码更具可读性和跨平台性。
2. 使用enumerate代替手动计数器，使代码更简洁。
3. 使用字典data来存储每次读取的DataFrame。
4. 在循环结束后，使用pd.concat一次性合并字典中的所有DataFrame。
5. memory_map=True, low_memory=False 参数可以帮助处理大型CSV文件。
6. .set_index('Case').squeeze() 将'Case'列设置为索引，并将DataFrame压缩成Series，以减少内存占用。
7. .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())

代码解释：

1. 定义一个read_csv函数，用于读取单个CSV文件。
2. 使用ThreadPoolExecutor创建线程池，max_workers参数指定线程数量。
3. 使用executor.map将read_csv函数应用到df的每一行，实现并行读取。
4. df[['File ID', 'File Name']].to_dict('records') 将DataFrame转换为字典列表，方便传递给read_csv函数。
5. enumerate 函数为每个文件添加一个计数器，用于打印进度。

注意事项：

• max_workers参数需要根据CPU核心数和实际情况进行调整。
• 多线程适用于I/O密集型任务，如果CPU计算是瓶颈，多线程可能无法带来显著的性能提升。
• 注意线程安全问题，避免多个线程同时访问共享资源。

总结

通过避免在循环中使用pd.concat，而是先将数据收集起来，最后一次性合并，可以显著提升处理大量CSV文件的效率。如果读取CSV文件是瓶颈，可以考虑使用多线程并行读取，进一步提升效率。在实际应用中，需要根据具体情况选择合适的优化方案。