高效处理大量CSV文件：Pandas循环优化与多线程应用-Python教程-PHP中文网

高效处理大量CSV文件：Pandas循环优化与多线程应用

霞舞

发布： 2025-11-18 13:45:01

原创

347人浏览过

高效处理大量csv文件：pandas循环优化与多线程应用

本文旨在解决在循环中处理大量CSV文件时遇到的性能瓶颈问题，重点介绍如何通过避免在循环中使用`concat`操作，以及利用Python字典和`pandas.concat`函数进行优化。此外，还探讨了使用多线程并行处理CSV文件以进一步提升效率的方法，并提供详细的代码示例和解释。

Pandas循环处理大量CSV文件的优化策略

在数据处理任务中，经常需要循环读取并处理大量的CSV文件。如果每个文件的数据量较大，且文件数量众多，那么循环中的某些操作可能会成为性能瓶颈，导致程序运行缓慢。其中，在循环中频繁使用pandas.concat函数就是一个常见的性能问题。

问题分析：为什么循环中的concat很慢？

pandas.concat函数用于将多个DataFrame对象沿着指定的轴进行连接。当在循环中调用concat时，每次迭代都会创建一个新的DataFrame对象，并将之前的结果复制到新的对象中。这种频繁的内存分配和数据复制操作会消耗大量的时间和资源，导致程序运行效率低下。

解决方案：避免循环中的concat

为了解决这个问题，可以采用以下策略：

将数据收集到Python字典中： 在循环中，将每个CSV文件读取并处理后的数据存储到Python字典中，其中键可以是文件名或其他标识符，值是对应的DataFrame或Series对象。
一次性concat： 在循环结束后，使用pandas.concat函数将字典中的所有DataFrame或Series对象一次性连接起来。

这种方法避免了在循环中频繁创建和复制DataFrame对象，从而显著提高了程序的运行效率。

代码示例：使用字典和concat优化循环

以下代码示例演示了如何使用字典和pandas.concat函数优化循环处理CSV文件的过程：

文心大模型

百度飞桨-文心大模型 ERNIE 3.0 文本理解与创作

查看详情

import pathlib
import pandas as pd

# 假设 root_path 是包含所有CSV文件的根目录
root_path = pathlib.Path('root')

# 创建一个DataFrame，其中包含文件ID和文件名
df = pd.DataFrame({'File ID': ['folderA', 'folderB'], 'File Name': ['file001.txt', 'file002.txt']})

data = {}
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)
    # 将 'Case' 列设置为索引，并将 DataFrame 转换为 Series
    data[folder_file_id] = file_data.set_index('Case').squeeze()
    print(count)

# 使用 pandas.concat 函数将字典中的所有 Series 对象连接起来
merged_data = (pd.concat(data, names=['folder_file_id'])
                 .unstack('Case').reset_index())

print(merged_data)

登录后复制

代码解释：

pathlib.Path 用于更方便地处理文件路径。
enumerate 函数用于在循环中同时获取索引和行数据。
file_data.set_index('Case').squeeze() 将 'Case' 列设置为索引，并将 DataFrame 转换为 Series，这可以简化后续的连接操作。
pd.concat(data, names=['folder_file_id']) 将字典 data 中的所有 Series 对象沿着索引连接起来，并使用 folder_file_id 作为索引的名称。
.unstack('Case') 将 'Case' 索引转换为列，.reset_index() 重置索引。

注意事项：

确保所有CSV文件具有相同的列结构，以便能够正确地连接它们。
根据实际情况调整pandas.read_csv函数的参数，例如sep（分隔符）、header（是否包含标题行）等。
如果CSV文件非常大，可以考虑使用chunksize参数分块读取文件，以减少内存占用。

多线程并行处理CSV文件

除了避免循环中的concat操作，还可以使用多线程并行处理CSV文件，以进一步提高程序的运行效率。多线程可以将任务分解成多个子任务，并同时执行这些子任务，从而缩短总的运行时间。

代码示例：使用多线程并行处理CSV文件

以下代码示例演示了如何使用concurrent.futures.ThreadPoolExecutor实现多线程并行处理CSV文件的过程：

from concurrent.futures import ThreadPoolExecutor
import pathlib
import pandas as pd

# 假设 root_path 是包含所有CSV文件的根目录
root_path = pathlib.Path('root')

# 创建一个DataFrame，其中包含文件ID和文件名
df = pd.DataFrame({'File ID': ['folderA', 'folderB'], 'File Name': ['file001.txt', 'file002.txt']})


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}'

    # 读取CSV文件，并指定列名和分隔符
    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())

print(merged_data)

登录后复制

代码解释：

concurrent.futures.ThreadPoolExecutor 用于创建线程池，管理和调度线程的执行。
max_workers 参数指定线程池中线程的最大数量。
executor.map(read_csv, batch) 将 read_csv 函数应用到 batch 中的每个元素，并返回一个迭代器，其中包含每个线程的返回值。
dict(data) 将迭代器转换为字典，其中键是文件名，值是对应的DataFrame或Series对象。