新闻中心

Pandas批量文件处理性能优化：避免循环内concat与并发实践

2025-11-18
浏览次数：次
返回列表

Pandas批量文件处理性能优化：避免循环内concat与并发实践

本文深入探讨了在pandas中高效处理和合并大量csv文件的方法。针对循环内部频繁使用`pd.concat`导致的性能瓶颈，文章提出了将数据收集到字典中并在循环结束后进行一次性合并的优化策略。此外，结合`pathlib`进行路径管理和利用多线程实现并发处理，进一步提升了数据处理效率和内存利用率，为大规模数据整合提供了专业的解决方案。

引言：循环内concat的性能陷阱

在数据处理工作中，我们经常需要从多个文件中读取数据并将其合并到一个大型Pandas DataFrame中。一个常见的直觉做法是在循环中逐个读取文件，然后使用pd.concat将每个文件的数据追加到主DataFrame。然而，当文件数量庞大（例如上千个）且每个文件的数据量不小（例如15MB，超过10,000行）时，这种做法会导致严重的性能问题。初始的几次循环可能很快完成，但随着主DataFrame的不断增大，每次concat操作所需的时间会呈指数级增长，最终使得整个过程变得异常缓慢，甚至可能耗尽系统内存。

原始问题与低效实践分析

考虑以下场景：我们有一个包含文件路径信息的DataFrame df，需要遍历其中的每一行，读取对应的CSV文件，对其进行转置和格式化，最终合并到一个名为 merged_data 的大型DataFrame中。

原始代码示例

import pandas as pd
import os

# 假设 root_path 和 df 已经定义
# root_path = '/path/to/your/root'
# df = pd.DataFrame({'File ID': ['folderA', 'folderB'], 'File Name': ['file001.txt', 'file002.txt']})

merged_data = pd.DataFrame()
count = 0
for index, row in df.iterrows():
    folder_name = row['File ID'].strip()
    file_name = row['File Name'].strip()
    file_path = os.path.join(root_path, folder_name, file_name)

    # 读取、转置并格式化文件数据
    file_data = pd.read_csv(file_path, names=['Case', f'{folder_name}_{file_name}'], sep='\t')
    file_data_transposed = file_data.set_index('Case').T.reset_index(drop=True)
    file_data_transposed.insert(loc=0, column='folder_file_id', value=str(folder_name+'_'+file_name))

    # 每次循环都进行 concat
    merged_data = pd.concat([merged_data, file_data_transposed], axis=0, ignore_index=True)
    count = count + 1
    print(count)

性能瓶颈分析

上述代码的性能问题主要源于以下几点：

  1. 循环内频繁 pd.concat: pd.concat操作在每次执行时，会创建一个新的DataFrame，并将现有数据和新数据复制到其中。随着merged_data的增大，每次复制的数据量也随之增加，导致内存分配和数据复制的开销急剧上升。这是导致性能呈指数级下降的主要原因。
  2. df.iterrows(): 虽然在小规模数据上不是主要瓶颈，但对于大型DataFrame，iterrows()会返回一个迭代器，每次迭代生成一个Series，这比向量化操作效率低。
  3. os.path.join: 虽然功能正确，但os.path模块在处理路径时不如pathlib模块面向对象且简洁。

优化方案一：延迟合并与Pathlib

解决循环内concat性能问题的核心思想是：避免在循环中重复执行昂贵的操作，而是将所有中间结果收集起来，在循环结束后一次性执行合并。

核心思想

我们将不再每次迭代都将数据追加到merged_data，而是将每个文件处理后的结果（通常是一个Pandas Series或DataFrame片段）存储在一个Python字典中。字典的键可以是文件的唯一标识符，值则是处理后的数据。循环结束后，我们再将这个字典传递给pd.concat，进行一次性高效合并。

优化代码实现

import pathlib
import pandas as pd

# 假设 root_path 和 df 已经定义
root_path = pathlib.Path('root') # 使用 pathlib 代替 os.path
df = pd.DataFrame({
    'File ID': ['folderA', 'folderB'], 
    'File Name': ['file001.txt', 'file002.txt']
})

data = {} # 用于收集所有处理后的数据
# 使用 enumerate 而非外部计数器，并直接迭代 df.iterrows()
for count, (_, row) in enumerate(df.iterrows(), 1):
    folder_name = row['File ID'].strip()
    file_name = row['File Name'].strip()

    # 使用 pathlib 构建文件路径，更简洁安全
    file_path = root_path / folder_name / file_name
    folder_file_id = f'{folder_name}_{file_name}'

    # 读取CSV文件，指定 header=None 因为文件没有表头
    # memory_map=True 可以提高大文件读取效率
    # low_memory=False 确保正确推断所有列的数据类型
    file_data = pd.read_csv(file_path, header=None, sep='\t',
                            names=['Case', folder_file_id],
                            memory_map=True, low_memory=False)

    # 设置 'Case' 列为索引，并使用 squeeze() 将单列DataFrame转换为Series
    data[folder_file_id] = file_data.set_index('Case').squeeze()
    print(count)

# 循环结束后，一次性进行 concat
merged_data = (pd.concat(data, names=['folder_file_id'])
                 .unstack('Case').reset_index())

示例输入数据：

Whimsical Whimsical

Whimsical推出的AI思维导图工具

Whimsical 182 查看详情 Whimsical 
# df
   File ID    File Name
0  folderA  file001.txt
1  folderB  file002.txt

# root/folderA/file001.txt
0   1234
1   5678
2   9012
3   3456
4   7890

# root/folderB/file002.txt
0   4567
1   8901
2   2345
3   6789

示例输出：

>>> merged_data
Case       folder_file_id       0       1       2       3       4
0     folderA_file001.txt  1234.0  5678.0  9012.0  3456.0  7890.0
1     folderB_file002.txt  4567.0  8901.0  2345.0  6789.0     NaN

代码解析与优势

  1. pathlib 模块: pathlib.Path对象提供了更直观、面向对象的方式来处理文件系统路径。例如，使用/运算符即可拼接路径，替代了os.path.join，代码更具可读性。
  2. 数据收集到字典 data: 这是核心优化点。每次循环将处理后的数据（一个Series）存储到字典中，避免了频繁的内存重分配和数据复制。
  3. pd.read_csv 参数优化:
    • header=None: 明确指定文件没有表头，避免Pandas误将第一行数据作为表头。
    • names=['Case', folder_file_id]: 直接为列指定名称。
    • memory_map=True: 尝试将文件映射到内存，对于大文件可能提高读取效率。
    • low_memory=False: 告诉Pandas在读取整个文件后才推断数据类型，这会消耗更多内存，但能避免混合类型列的问题，尤其在有大量列时非常有用。
  4. set_index('Case').squeeze(): set_index('Case')将'Case'列设为索引。.squeeze()方法用于移除单维度的条目，例如将一个只有一列的DataFrame转换为一个Series，这对于后续的concat和unstack操作非常方便。
  5. 一次性 pd.concat: 循环结束后，将包含所有Series的字典data传递给pd.concat。Pandas会高效地将这些Series合并成一个DataFrame。
    • names=['folder_file_id']: 为新生成的层级索引指定名称。
  6. unstack('Case').reset_index():
    • pd.concat(data, names=['folder_file_id']) 会生成一个多层索引的Series，其中第一层索引是folder_file_id，第二层是Case。
    • .unstack('Case') 会将Case索引级别转换为列，实现数据的“宽”格式转换，使其符合原始需求中的转置效果。
    • .reset_index() 将多层索引转换为普通列，并重置数字索引，得到最终的扁平化DataFrame。

优化方案二：利用多线程并发处理

对于I/O密集型任务（如读取大量文件），即使优化了concat，文件读取本身仍可能成为瓶颈。在这种情况下，可以考虑使用多线程（或多进程）来并行化文件读取过程。Python的concurrent.futures模块提供了方便的接口来实现这一点。

并发处理的原理

多线程在Python中受GIL（全局解释器锁）的限制，对于CPU密集型任务效果不佳。但对于I/O密集型任务（如文件读写、网络请求），当一个线程等待I/O操作完成时，GIL会被释放，允许其他线程执行Python代码。因此，多线程可以显著提高I/O密集型任务的整体吞吐量。

多线程代码实现

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

root_path = pathlib.Path('root')
df = pd.DataFrame({
    'File ID': ['folderA', 'folderB'], 
    'File Name': ['file001.txt', 'file002.txt']
})

def read_and_process_csv(args):
    """
    一个辅助函数，用于在单独的线程中读取和处理单个CSV文件。
    """
    count, row_dict = args # 解包参数，row_dict 是 df.to_dict('records') 的一行
    folder_name = row_dict['File ID'].strip()
    file_name = row_dict['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(f"Processing {count}: {folder_file_id}")
    return folder_file_id, file_data.set_index('Case').squeeze()

# 创建一个线程池，max_workers 根据CPU核心数和I/O负载调整
with ThreadPoolExecutor(max_workers=4) as executor: # 示例使用4个工作线程
    # 将 DataFrame 转换为字典列表，以便传递给线程池
    # enumerate 用于添加计数器
    batch_args = enumerate(df[['File ID', 'File Name']].to_dict('records'), 1)

    # 使用 executor.map 并行执行 read_and_process_csv 函数
    # data 将是一个迭代器，按提交顺序返回结果
    data_iterator = executor.map(read_and_process_csv, batch_args)

    # 将迭代器转换为字典，以便 pd.concat 处理
    data_dict = dict(data_iterator)

# 循环结束后，一次性进行 concat
merged_data = (pd.concat(data_dict, names=['folder_file_id'])
                 .unstack('Case').reset_index())

适用场景与注意事项

  1. ThreadPoolExecutor: 创建一个线程池，max_workers参数控制同时运行的最大线程数。对于I/O密集型任务，通常可以设置得比CPU核心数高一些。
  2. read_and_process_csv 函数: 这个函数封装了单个文件的读取和处理逻辑，它将作为线程池的任务被执行。
  3. df.to_dict('records'): 将DataFrame转换为一个字典列表，每个字典代表一行数据。这样可以方便地将行数据作为参数传递给线程函数。
  4. executor.map(): 这是ThreadPoolExecutor提供的一个高阶函数，它将一个函数应用到可迭代对象的每个元素上，并返回一个迭代器，其中包含函数调用的结果。它会按照提交的顺序返回结果，这对于后续的pd.concat非常重要。
  5. I/O密集型任务: 多线程特别适合文件读取这种I/O密集型任务。如果任务是CPU密集型的（例如大量的数值计算），则应考虑使用ProcessPoolExecutor（多进程）来规避GIL的限制。
  6. 错误处理: 在生产环境中，需要为并发任务添加适当的错误处理机制，例如使用try-except块捕获文件读取或处理中的异常。
  7. 内存管理: 尽管并发读取可以加快速度，但如果每个文件都很大，所有文件的数据最终仍会加载到内存中。因此，需要确保系统有足够的内存来容纳所有合并后的数据。

总结与最佳实践

在Pandas中高效处理和合并大量文件是数据工程中的常见挑战。本文提供了两种关键的优化策略：

  1. 延迟合并: 永远不要在循环内部频繁地使用pd.concat。相反，将每个迭代产生的数据片段收集到一个列表或字典中，然后在循环结束后执行一次性的大规模合并。这能显著减少内存分配和数据复制的开销。
  2. 利用pathlib进行路径管理: pathlib模块提供了更现代、更直观、更安全的路径操作方式，推荐替代os.path。
  3. 并发处理: 对于I/O密集型任务，可以利用concurrent.futures.ThreadPoolExecutor实现多线程并发读取和初步处理，进一步缩短整体执行时间。

通过采纳这些最佳实践，您可以有效地处理数千甚至数万个文件，将原本耗时数小时甚至数天的任务缩短到可接受的时间范围内，极大地提升数据处理效率。在实际应用中，根据具体的数据量、文件大小和系统资源，可以灵活选择并组合这些优化方法。

以上就是Pandas批量文件处理性能优化：避免循环内concat与并发实践的详细内容，更多请关注其它相关文章！


# 创建一个  # 天津发展网站建设经历  # se01 seo  # 广州seo优化教程  # 廊坊网站建设程序  # 包装网站推广方法大全  # 天津seo培训哪家好  # 营销网站建设方案咨询  # 高档模型网站推广方案  # 敦煌网平台如何推广营销  # 山西家居网站建设平台  # 它将  # 运算符  # python  # 数据处理  # 面向对象  # 这是  # 结束后  # 转换为  # 迭代  # 多线程  # 可迭代对象  # csv文件  # 性能瓶颈  # csv 


相关栏目： 【 科技资讯46185 】 【 网络学院92790 】


相关推荐： 凉拌黄瓜怎么拌更入味 凉拌黄瓜简单家常做法  提升屏幕阅读器对“m”时间单位的播报准确性：HTML与CSS组合解决方案  Python类型检查：优化关联可选属性的Mypy推断策略  css滚动区域卡顿如何改善_css滚动问题用will-change优化渲染  php源码怎么看淘宝客系统_看php源码淘宝客系统技巧  c++如何使用Catch2编写单元测试_c++简洁易用的BDD风格测试框架  css子元素高度不一致导致布局错位怎么办_使用align-items:stretch解决高度差异  Win11怎么设置鼠标主按键_Win11鼠标左右键功能互换  谷歌google账号注册详细步骤 谷歌账号注册官方教程  创客贴用户入口官网登录 创客贴网页版电脑版系统  PostgreSQL海量数据高效导入策略：Python与Django实践指南  Descript怎样用AI剪辑自动去噪_Descript用AI剪辑自动去噪【自动降噪】  微信客户端如何收红包_微信客户端接收红包使用教程  抖音隐秘迷城小游戏入口_ 抖音冒险解谜小游戏秒玩  微信怎么把收藏的内容分类管理 微信收藏内容标签分类方法  如何在Promise链中有效终止错误处理后的执行  c++中的const_cast和reinterpret_cast怎么用_c++四种类型转换  Mac怎么锁定备忘录_Mac备忘录加密设置教程  最新韩小圈网页版登录入口_官网在线观看官方链接  俄罗斯浏览器官网直达链接 俄罗斯浏览器最新在线入口导航  AI泡沫首次被“刺破”：GPU十年都无法存活！  在J*a中如何使用BigDecimal进行高精度计算_BigDecimal类应用指南  c++项目目录结构应该如何组织_c++工程化项目结构规范  天眼查怎么看公司融资情况 天眼查企业融资历史查询步骤【攻略】  黑猫投诉统一入口官网 消费者权益保护投诉平台  Lar*el 递归关系中排除指定分支的教程  J*a递归快速排序中静态变量的状态管理与陷阱  夸克浏览器网页版最新地址 夸克浏览器官方入口合集  支付宝如何设置安全保护_支付宝安全设置的全面教程  漫蛙2网页版漫画入口 漫蛙漫画在线官方登录  J*a中实现Go语言select通道多路复用机制  c++20的std::jthread是什么_c++可中断线程与RAII式管理  Go语言中Map存储的结构体如何调用指针方法：深入解析与实践  Linux如何排查内存不足OOME问题_LinuxOOM分析教程  德邦快递查询平台 德邦快递物流信息查询入口  响应式图片在网页设计中的正确实现方法  React/Next.js中实现列表项的动态选择与移动  Golang如何处理RPC请求负载均衡_Golang RPC请求负载均衡策略与实践  漫蛙manwa官网登录界面_漫蛙漫画网页版主站入口  如何优雅地扩展SprykerGlue后端API授权逻辑，使用spryker/glue-backend-api-application-authorization-connector-extension  vivo浏览器怎么扫描二维码 vivo浏览器内置扫一扫功能使用方法  Lar*el头像管理：图片缩放与旧文件删除的最佳实践  深入理解字体排版：Adobe光学字偶距与CSS字偶距的差异与实现  React Router v6 教程：构建认证保护的私有路由与重定向策略  移动端XML文件怎么转换成Excel 手机和平板上的解决方案  AO3官方镜像站点汇总 AO3同人作品网页版直达链接  怎样把文件彻底粉碎无法恢复_Windows下安全删除敏感数据【隐私保护】  C++如何解决segmentation fault_C++段错误调试与原因分析  蛙漫漫画官网在线入口 蛙漫全本漫画免费阅读平台  钉钉视频会议声音异常如何处理 钉钉会议音频修复技巧 

上一篇：如何在Golang中实现错误回调机制
下一篇：解决Jupyter Notebook新版本中无法直接粘贴文本的问题

搜索