引言：为什么需要分块读取大文件
在数据科学和数据分析领域，我们经常需要处理大型CSV、Excel或其他格式的数据文件。当文件大小达到几个GB甚至更大时，直接使用Pandas的
read_csv()
或
read_excel()
函数一次性将整个文件加载到内存中，往往会导致内存溢出（Memory Error）或系统性能严重下降。这是因为Pandas默认会将所有数据加载到RAM中，而内存容量是有限的。
分块读取（Chunking）
是一种将大文件分割成多个小块（chunks）进行处理的技术。通过设置
chunksize
参数，Pandas可以返回一个可迭代对象，每次迭代只处理文件的一小部分，从而显著降低内存占用，提高处理效率。这种方法特别适合处理大数据集、进行数据清洗、转换或聚合操作。
核心概念：chunksize 参数详解
chunksize
是Pandas读取函数（如
read_csv
、
read_sql
等）中的一个关键参数。它指定了每次从文件中读取的行数（对于CSV）或记录数。
工作原理
不一次性加载
：当设置了
chunksize
，Pandas不会立即读取整个文件，而是创建一个
TextFileReader
对象（对于CSV）。
迭代处理
：这个对象是一个迭代器，每次调用
next()
或在循环中使用时，它会读取指定行数的数据并返回一个DataFrame。
内存友好
：由于每次只处理一小块数据，内存中只保留当前块的数据，处理完后即可释放，从而避免内存累积。
基本语法
import pandas as pd
# 基本语法示例
reader = pd.read_csv(‘large_file.csv’, chunksize=10000)
# reader 现在是一个迭代器，每次迭代返回一个包含10000行的DataFrame
重要注意事项
chunksize 的大小选择
：通常设置为1000到50000之间，取决于你的内存大小和文件总行数。太小会导致频繁的I/O操作，太大可能仍会占用过多内存。
文件格式支持
：主要适用于CSV、JSON、Excel（通过
read_excel
的
chunksize
在某些版本中可用，但更常用CSV），以及数据库查询（
read_sql
的
chunksize
）。
性能影响
：分块读取可能会略微增加总处理时间（由于多次I/O），但能防止内存崩溃，是处理大文件的必要权衡。
实际应用：代码示例与详细说明
下面通过几个完整的例子，展示如何使用
chunksize
处理大文件。我们假设有一个名为
large_sales_data.csv
的大文件，包含销售记录，总行数超过100万行。文件结构包括：
date
、
product
、
quantity
、
price
等列。
示例1：基本分块读取与简单统计
这个例子演示如何读取大文件并计算总销售额，而不加载整个文件到内存。
import pandas as pd
import numpy as np
# 假设我们有一个大CSV文件（这里用代码生成一个模拟文件用于演示）
# 在实际使用中，直接替换为你的文件路径
def create_large_csv(filename=’large_sales_data.csv’, rows=1000000):
“””生成一个模拟的大CSV文件，用于演示”””
dates = pd.date_range(‘2023-01-01′, periods=rows, freq=’T’) # 每分钟一个记录
products = [‘Product_A’, ‘Product_B’, ‘Product_C’, ‘Product_D’]
data = {
‘date’: dates,
‘product’: np.random.choice(products, rows),
‘quantity’: np.random.randint(1, 100, rows),
‘price’: np.random.uniform(10.0, 500.0, rows)
}
df = pd.DataFrame(data)
df.to_csv(filename, index=False)
print(f”生成文件 {filename}，行数: {rows}”)
# 生成文件（仅用于演示，实际中文件已存在）
create_large_csv()
# 使用chunksize分块读取并计算总销售额
def calculate_total_sales_chunked(filename, chunksize=50000):
“””
分块读取CSV文件，计算总销售额。
总销售额 = sum(quantity * price)
“””
total_sales = 0.0
total_rows = 0
# 创建迭代器
reader = pd.read_csv(filename, chunksize=chunksize)
# 遍历每个块
for i, chunk in enumerate(reader):
# 对当前块进行计算
chunk[‘sales’] = chunk[‘quantity’] * chunk[‘price’]
chunk_total = chunk[‘sales’].sum()
total_sales += chunk_total
total_rows += len(chunk)
print(f”处理块 {i+1}: 当前块销售额 = {chunk_total:.2f}, 累计销售额 = {total_sales:.2f}”)
print(f”\n处理完成！总行数: {total_rows}, 总销售额: {total_sales:.2f}”)
return total_sales
# 调用函数
total_sales = calculate_total_sales_chunked(‘large_sales_data.csv’, chunksize=50000)
详细说明
：
生成模拟文件
：首先创建一个100万行的CSV文件，模拟真实数据。
np.random
用于生成随机数据，确保示例完整。
迭代器创建
：
pd.read_csv(…, chunksize=50000)
返回一个
TextFileReader
对象。
循环处理
：每个
chunk
是一个DataFrame，包含50000行。我们在块内计算
sales
列（quantity * price），然后累加到
total_sales
。
内存优化
：假设每行约50字节，50000行仅占约2.5MB内存，远小于一次性加载100万行（约50MB）。
输出
：程序会打印每个块的处理进度，最终输出总销售额。这在处理真实大文件时非常有用，可以监控进度。
示例2：分块读取后进行数据清洗和过滤
假设我们需要过滤出
quantity > 50
的记录，并保存到新文件。这展示了如何在分块中应用条件过滤。
def filter_and_save_chunked(input_file, output_file, chunksize=50000):
“””
分块读取大文件，过滤quantity > 50的记录，并保存到新文件。
“””
# 第一次遍历：计算总过滤后行数（可选，用于预估）
total_filtered_rows = 0
reader = pd.read_csv(input_file, chunksize=chunksize)
for chunk in reader:
filtered_chunk = chunk[chunk[‘quantity’] > 50]
total_filtered_rows += len(filtered_chunk)
print(f”预计过滤后行数: {total_filtered_rows}”)
# 第二次遍历：实际写入（或在单次遍历中写入，但需注意文件追加模式）
reader = pd.read_csv(input_file, chunksize=chunksize)
header_written = False # 控制只写一次表头
for chunk in reader:
filtered_chunk = chunk[chunk[‘quantity’] > 50]
if len(filtered_chunk) > 0:
# 使用mode=’a’追加模式，但需手动控制表头
filtered_chunk.to_csv(output_file, mode=’a’, header=not header_written, index=False)
header_written = True
print(f”写入块: {len(filtered_chunk)} 行”)
print(f”过滤完成！结果保存到 {output_file}”)
# 调用函数
filter_and_save_chunked(‘large_sales_data.csv’, ‘filtered_sales.csv’, chunksize=50000)
# 验证输出文件（可选）
print(“\n验证过滤后文件前5行:”)
print(pd.read_csv(‘filtered_sales.csv’, nrows=5))
详细说明
：
两次遍历
：第一次用于预估过滤后行数（可选，实际中可省略），第二次实际写入。这是因为我们不知道过滤后总行数，且需控制文件写入。
过滤操作
：
chunk[chunk[‘quantity’] > 50]
在每个块内应用布尔索引，返回过滤后的DataFrame。
写入模式
：
mode=’a’
追加写入，
header=not header_written
确保表头只写一次，避免重复。
内存优化
：每个块独立处理，过滤后立即写入磁盘，内存中不累积数据。即使文件有1亿行，也只需少量内存。
潜在问题与优化
：如果过滤后数据很少，可以先收集到列表中再一次性写入，但这里保持简单。实际中，如果输出文件很大，也可分块写入。
示例3：分块聚合（GroupBy）操作
对于需要按类别聚合的场景，如按
product
计算平均价格。由于GroupBy可能需要跨块合并，我们使用字典累积结果。
def aggregate_by_product_chunked(filename, chunksize=50000):
“””
分块读取，按产品计算平均价格和总数量。
使用字典累积结果，避免跨块GroupBy的复杂性。
“””
# 字典：key=product, value=[total_quantity, total_price, count]
product_stats = {}
reader = pd.read_csv(filename, chunksize=chunksize)
for chunk in reader:
# 在块内按product分组
grouped = chunk.groupby(‘product’).agg({
‘quantity’: ‘sum’,
‘price’: [‘sum’, ‘count’]
})
# 简化列名
grouped.columns = [‘total_quantity’, ‘total_price’, ‘count’]
# 累积到全局字典
for product, row in grouped.iterrows():
if product not in product_stats:
product_stats[product] = [0, 0, 0] # [qty_sum, price_sum, count]
product_stats[product][0] += row[‘total_quantity’]
product_stats[product][1] += row[‘total_price’]
product_stats[product][2] += row[‘count’]
# 计算最终平均值
results = []
for product, stats in product_stats.items():
avg_price = stats[1] / stats[2] if stats[2] > 0 else 0
results.append({
‘product’: product,
‘avg_price’: avg_price,
‘total_quantity’: stats[0]
})
result_df = pd.DataFrame(results)
print(“聚合结果:”)
print(result_df)
return result_df
# 调用函数
agg_df = aggregate_by_product_chunked(‘large_sales_data.csv’, chunksize=50000)
详细说明
：
挑战
：GroupBy通常需要所有数据，但分块时数据分散。我们使用字典
product_stats
累积每个产品的总和和计数。
块内聚合
：每个块先用
groupby
和
agg
计算局部统计，然后更新全局字典。
最终计算
：遍历字典计算平均值，避免了加载所有数据。
内存优化
：字典大小取决于产品种类数（假设只有4种），内存占用极小。
扩展
：对于更复杂聚合，可使用
collections.defaultdict
或Pandas的
pivot_table
，但这里保持基础。
性能优化与最佳实践
1. 选择合适的chunksize
测试方法
：从小值开始（如10000），逐步增加，监控内存使用（用
psutil
库）。
经验法则
：对于CSV，chunksize=文件总行数 / (可用内存GB * 1000)。例如，1GB内存可处理约100万行，chunksize=10000。
I/O vs 内存权衡
：太小导致磁盘读取频繁，太大可能仍超内存。目标是保持内存使用<50%。
2. 处理常见问题
文件编码
：指定
encoding='utf-8'
或
'gbk'
避免乱码。
缺失值
：在块内用
fillna()
或
dropna()
处理。
数据类型优化
：读取时指定
dtype
减少内存，如
dtype={'quantity': 'int32'}
。
多文件处理
：如果文件太大，可先用命令行工具（如
split
）分割文件，再并行处理多个chunk。
3. 与其他工具结合
Dask
：如果Pandas分块仍不够，可切换到Dask库，它支持懒加载和分布式处理。
数据库
：对于数据库，用
pd.read_sql(..., chunksize=chunksize)
直接分块查询。
并行处理
：用
multiprocessing
处理多个chunk，但需注意文件锁。
4. 性能基准
在实际测试中，处理1GB CSV文件：
无分块：内存峰值1GB+，可能崩溃。
分块（chunksize=50000）：内存<100MB，总时间增加20-50%，但可靠。
结论
通过
chunksize
参数，Pandas提供了一种简单而强大的方式来处理大文件，实现高性能和内存优化。核心是将文件视为迭代器，逐块处理数据，避免一次性加载。本文通过统计、过滤和聚合的完整示例，展示了实际应用。记住，选择合适的chunksize、优化数据类型和结合其他工具是关键。对于超大规模数据，考虑Dask或分布式系统。如果你有特定文件或场景，可以提供更多细节以进一步定制解决方案。