Pandas大规模数据分块读取与内存优化的实战指南

import pandas as pd

# 错误示范1：直接读取整个文件
df = pd.read_csv('large_file.csv')  # 内存占用飙升至9.2GB，程序崩溃

# 错误示范2：未指定数据类型
df = pd.read_csv('large_file.csv', dtype=object)  # 内存占用翻倍

内存占用(MB) ≈ 行数 × 列数 × 每个值的字节数 / 1024²

# 分块读取示例：每次处理10万行
chunk_size = 100000
chunks = pd.read_csv('sales_data.csv', chunksize=chunk_size)

results = []
for chunk in chunks:
    # 对每个数据块进行处理
    chunk_processed = chunk[chunk['amount'] > 1000]
    results.append(chunk_processed)

# 合并结果（注意内存控制）
final_df = pd.concat(results, ignore_index=True)

with pd.HDFStore('output.h5', mode='w') as store:
    for i, chunk in enumerate(pd.read_csv('big_data.csv', chunksize=50000)):
        # 数据清洗逻辑
        cleaned = chunk.dropna(subset=['price'])
        store.append(f'chunk_{i}', cleaned, index=False)

import pandas as pd
from collections import defaultdict

user_stats = defaultdict(lambda: {'count':0, 'duration':0})

for chunk in pd.read_csv('access_logs.csv', chunksize=100000):
    for _, row in chunk.iterrows():
        user = row['user_id']
        user_stats[user]['count'] += 1
        user_stats[user]['duration'] += row['duration']

# 转换为DataFrame
result_df = pd.DataFrame.from_dict(user_stats, orient='index')

# 原始读取（自动推断类型，可能不最优）
df = pd.read_csv('data.csv')  # 内存占用：1.2GB

# 优化后读取（指定精确类型）
dtypes = {
    'id': 'int32',
    'price': 'float32',
    'category': 'category',  # 分类数据专用类型
    'date': 'datetime64[ns]'
}
df_optimized = pd.read_csv('data.csv', dtype=dtypes)  # 内存占用：480MB

# 原始字符串列（占用大）
df['product_type'] = ['A','B','A','C'...]  # 每个值重复存储

# 转换为分类类型（节省内存）
df['product_type'] = df['product_type'].astype('category')

# 进一步优化：使用数值编码
df['product_code'] = df['product_type'].cat.codes

# 原始缺失值（NaN占用空间）
df = pd.DataFrame({'A': [1, None, 3], 'B': [None, 'x', None]})

# 优化方案1：用特定值填充（适合数值列）
df['A'].fillna(0, inplace=True)

# 优化方案2：用更紧凑的类型存储（适合字符串列）
df['B'] = df['B'].astype('category')

# 优化方案3：直接删除（当缺失比例高时）
df.dropna(subset=['important_column'], inplace=True)

# 创建稀疏DataFrame（大部分值为0或空）
import numpy as np
import pandas as pd

data = np.random.choice([0, 1], size=(1000000, 100), p=[0.99, 0.01])
df = pd.DataFrame(data)

# 转换为稀疏格式（节省95%内存）
sparse_df = df.astype(pd.SparseDtype("int8", 0))

# 原始读取（自动转为datetime64[ns]）
df = pd.read_csv('transactions.csv', parse_dates=['date'])  # 8字节/值

# 优化方案1：使用整数时间戳
df['timestamp'] = pd.to_datetime(df['date']).astype(np.int64) // 10**9  # 4字节/值

# 优化方案2：分离年月日（当不需要完整时间时）
df['year'] = pd.to_datetime(df['date']).dt.year  # int16
df['month'] = pd.to_datetime(df['date']).dt.month  # int8

# 保存为Parquet格式（压缩比高）
df.to_parquet('data.parquet', compression='snappy')

# 读取Parquet文件
df_parquet = pd.read_parquet('data.parquet')

# 识别高内存对象列
def memory_usage(df):
    return df.memory_usage(deep=True).sort_values(ascending=False)

# 对象列优化方案
for col in df.select_dtypes(include=['object']):
    # 尝试转换为category
    if df[col].nunique() / len(df) < 0.5:
        df[col] = df[col].astype('category')
    # 尝试转换为更紧凑的字符串表示
    elif df[col].str.len().max() < 50:
        pass  # 保持现状或考虑数值编码
    else:
        # 分割字符串或提取关键信息
        df[['part1','part2']] = df[col].str.split('|', expand=True)

import pandas as pd
import numpy as np

# 定义数据类型
dtypes = {
    'order_id': 'int64',
    'user_id': 'int32',
    'product_id': 'int32',
    'quantity': 'int16',
    'price': 'float32',
    'order_time': 'datetime64[ns]'
}

# 分块处理函数
def process_chunk(chunk):
    # 数据清洗
    chunk = chunk[chunk['price'] > 0]
    chunk = chunk[chunk['quantity'] > 0]
    
    # 特征工程
    chunk['total_amount'] = chunk['price'] * chunk['quantity']
    chunk['day_of_week'] = chunk['order_time'].dt.dayofweek
    
    # 按用户分组统计
    user_stats = chunk.groupby('user_id').agg({
        'total_amount': 'sum',
        'quantity': 'sum',
        'order_id': 'count'
    }).rename(columns={'order_id': 'order_count'})
    
    return user_stats

# 主处理流程
chunk_size = 500000
results = []

for i, chunk in enumerate(pd.read_csv(
    'orders_2020-2023.csv',
    dtype=dtypes,
    parse_dates=['order_time'],
    chunksize=chunk_size
)):
    print(f"Processing chunk {i+1}")
    results.append(process_chunk(chunk))

# 合并结果
final_result = pd.concat(results).groupby('user_id').sum()
final_result.to_parquet('user_stats.parquet')

预计内存占用(GB) = 行数 × 列数 × 8字节 / 1024³

from multiprocessing import Pool

def parallel_process(chunk):
    # 处理逻辑同前
    return process_chunk(chunk)

if __name__ == '__main__':
    chunks = pd.read_csv('big_data.csv', chunksize=100000)
    
    with Pool(processes=4) as pool:  # 使用4个CPU核心
        results = pool.map(parallel_process, chunks)
    
    final_result = pd.concat(results)