Pandas处理大CSV：向量化提升性能技巧

理解Pandas性能瓶颈：为何避免iterrows()和apply()

在处理大型数据集时，Python Pandas因其高效的数据结构和丰富的操作函数而广受欢迎。然而，许多初学者在使用Pandas时常会遇到性能问题，尤其是在尝试逐行处理数据时。一个常见的误区是使用DataFrame.iterrows()或DataFrame.apply()方法。

例如，原始问题中提供的伪代码片段展示了典型的逐行迭代模式：

import os
import pandas as pd

# 假设 dados 是一个已加载的DataFrame
# dados = pd.read_csv('your_file.csv')

for i, row in dados.iterrows():
    # 对每一行执行复杂逻辑
    # 例如：检查 row['column_a'] 或 row['column_b'] 是否包含特定值
    # 如果满足条件，将 row['column_c'] 添加到列表中
    pass

这种模式对于小型数据集可能可行，但当数据量达到数千行时，性能会急剧下降。根据原始问题描述，1000行数据需要约40秒，10000行则需要400秒，这表明处理百万行数据将耗费数小时甚至更长时间，效率极低。

核心原因在于： iterrows()和apply()本质上是Python级别的循环。Pandas底层是基于NumPy和C语言实现的，其大部分操作都经过高度优化。当使用Python循环逐行处理时，Pandas的这些底层优化优势无法发挥，每次迭代都需要将数据从优化的C结构转换回Python对象，导致巨大的开销。这使得这些方法在性能上远不如Pandas的内置向量化操作。

核心优化策略：拥抱向量化操作

Pandas的核心优势在于其能够对整个Series（列）或DataFrame执行操作，而无需显式的Python循环。这种方法称为“向量化操作”，它利用了NumPy数组的强大功能和C语言的执行效率。

什么是向量化操作？ 向量化操作是指将一个函数或操作应用到整个数组或Series上，而不是逐个元素地进行。Pandas会将这些操作转化为底层的C或NumPy函数调用，从而实现极高的执行速度。

示例：简单的向量化操作

考虑一个简单的例子，为DataFrame添加一个新列，该列是现有列加1的结果：

import pandas as pd

df = pd.DataFrame({'existing_column': [1, 2, 3, 4, 5]})

# 传统循环（应避免）
# new_column_list = []
# for val in df['existing_column']:
#     new_column_list.append(val + 1)
# df['new_column'] = new_column_list

# 向量化操作
df['new_column'] = df['existing_column'] + 1
print(df)

输出：

   existing_column  new_column
0                1           2
1                2           3
2                3           4
3                4           5
4                5           6

这里的df['existing_column'] + 1就是典型的向量化操作，它比任何Python循环都要快得多。

将原始问题逻辑转化为向量化操作

回到原始问题中的伪代码逻辑： “对于列表中的每个项，检查row[column_a]或row[column_b]是否包含该项作为值，如果为真，则将row[column_c]添加到列表中。”

假设我们有一个search_items列表，并且我们想检查column_a或column_b中的值是否精确匹配search_items中的任何一个。我们可以这样进行向量化：

import pandas as pd

# 模拟数据
data = {
    'column_a': ['apple', 'banana', 'orange', 'grape', 'kiwi'],
    'column_b': ['red', 'yellow', 'green', 'purple', 'yellow'],
    'column_c': [10, 20, 30, 40, 50]
}
dados = pd.DataFrame(data)

# 待搜索的项列表
search_items = ['banana', 'green', 'kiwi']

# 创建布尔掩码：检查 'column_a' 中的值是否在 search_items 中
mask_a = dados['column_a'].isin(search_items)

# 创建布尔掩码：检查 'column_b' 中的值是否在 search_items 中
mask_b = dados['column_b'].isin(search_items)

# 合并两个掩码：如果 column_a 或 column_b 满足条件
combined_mask = mask_a | mask_b

# 使用合并后的掩码选择 'column_c' 中的值，并转换为列表
result_list = dados.loc[combined_mask, 'column_c'].tolist()

print("符合条件的 column_c 值列表:", result_list)

输出：

符合条件的 column_c 值列表: [20, 30, 50]

这个向量化方法避免了显式循环，利用Pandas和NumPy的底层优化，大大提高了处理速度。对于字符串包含匹配（而非精确匹配），可以使用str.contains()结合正则表达式。

进阶优化：分块读取大型文件 (chunksize)

即使使用了向量化操作，如果CSV文件极其庞大（例如，数亿行或数十GB），一次性加载到内存中仍然可能导致内存溢出。在这种情况下，可以使用pd.read_csv()的chunksize参数来分块读取文件。

chunksize参数允许你指定每次读取的行数，read_csv会返回一个迭代器，每次迭代产生一个DataFrame块。你可以对每个块应用向量化操作，然后将结果聚合起来。

import pandas as pd
import os

# 假设文件路径
# 在实际应用中，请替换为你的CSV文件路径
desktop_path = os.path.join(os.path.join(os.environ["USERPROFILE"]), "Desktop")
file_path = os.path.join(desktop_path, 'your_large_file.csv') # 请替换为你的实际文件

# 模拟一个大型文件处理场景
# 待搜索的项列表
search_items = ['banana', 'green', 'kiwi']

# 定义一个函数来处理每个数据块
def process_chunk(chunk_df, search_items_list):
    mask_a = chunk_df['column_a'].isin(search_items_list)
    mask_b = chunk_df['column_b'].isin(search_items_list)
    combined_mask = mask_a | mask_b
    return chunk_df.loc[combined_mask, 'column_c'].tolist()

all_results = []
# 设置 chunksize，例如每次读取100,000行
# 根据你的内存和文件大小调整此值
chunk_size = 100000

# 迭代读取CSV文件
for chunk_id, chunk in enumerate(pd.read_csv(file_path, chunksize=chunk_size)):
    print(f"正在处理第 {chunk_id + 1} 个数据块...")
    chunk_result = process_chunk(chunk, search_items)
    all_results.extend(chunk_result)

print("\n所有符合条件的 column_c 值列表 (分块处理):", all_results)

通过分块处理，即使文件大小超过可用内存，也能有效地进行数据处理。这种方法适用于需要聚合结果或在每个块上独立完成任务的场景。

总结与最佳实践

处理大型CSV文件时，提高Pandas性能的关键在于：

1. 避免Python循环： 尽量不使用iterrows()、itertuples()或apply()（尤其是当apply需要对每一行执行复杂逻辑时）。它们会将Pandas的性能优势降至最低。
2. 优先使用向量化操作： 尽可能将数据操作转化为对整个Series或DataFrame的函数调用。利用Pandas内置的函数、NumPy函数和布尔索引。
3. 利用chunksize处理超大文件： 当文件大小超出内存限制时，使用pd.read_csv(chunksize=...)分块读取和处理数据。
4. 数据类型优化： 确保DataFrame中的列使用最合适的数据类型（例如，使用category类型处理重复的字符串，使用更小的整数类型）。
5. 考虑其他库： 对于真正海量的数据集（例如，TB级别），可以考虑使用Dask（分布式Pandas）或Polars（Rust-based，速度极快）等专门为大数据设计的库。

通过遵循这些最佳实践，你可以显著提高Python Pandas处理大型CSV文件的效率，从而在面对百万级甚至更大数据量时也能游刃有余。