【AI Study】第四天，Pandas（6）- 性能优化

文章概要

本文详细介绍 Pandas 的性能优化技术，包括：

• 内存优化
• 计算优化
• 大数据处理
• 实际应用示例

内存优化

数据类型优化

# 查看数据类型
df.dtypes

# 查看内存使用情况
df.memory_usage(deep=True)

# 优化数值类型
# 将 float64 转换为 float32
df['float_column'] = df['float_column'].astype('float32')

# 将 int64 转换为 int32 或 int16
df['int_column'] = df['int_column'].astype('int32')

# 优化分类数据
df['category_column'] = df['category_column'].astype('category')

# 优化日期时间
df['datetime_column'] = pd.to_datetime(df['datetime_column'])

内存使用分析

# 查看每列的内存使用
def memory_usage_by_column(df):
    return df.memory_usage(deep=True).sort_values(ascending=False)

# 查看数据类型分布
def dtype_distribution(df):
    return df.dtypes.value_counts()

# 查看空值比例
def null_ratio(df):
    return df.isnull().sum() / len(df)

# 内存使用分析报告
def memory_analysis_report(df):
    print("内存使用情况：")
    print(memory_usage_by_column(df))
    print("\n数据类型分布：")
    print(dtype_distribution(df))
    print("\n空值比例：")
    print(null_ratio(df))

内存清理

# 删除不需要的列
df = df.drop(['unused_column1', 'unused_column2'], axis=1)

# 删除重复行
df = df.drop_duplicates()

# 重置索引
df = df.reset_index(drop=True)

# 清理内存
import gc
gc.collect()

# 使用 inplace 操作
df.dropna(inplace=True)
df.fillna(0, inplace=True)

计算优化

向量化操作

# 避免循环，使用向量化操作
# 不推荐
for i in range(len(df)):
    df.loc[i, 'new_column'] = df.loc[i, 'column1'] + df.loc[i, 'column2']

# 推荐
df['new_column'] = df['column1'] + df['column2']

# 使用 apply 而不是循环
# 不推荐
for i in range(len(df)):
    df.loc[i, 'new_column'] = some_function(df.loc[i, 'column'])

# 推荐
df['new_column'] = df['column'].apply(some_function)

# 使用向量化函数
df['new_column'] = np.where(df['column'] > 0, 'positive', 'negative')

并行计算

# 使用 multiprocessing 进行并行计算
from multiprocessing import Pool

def process_chunk(chunk):
    # 处理数据块的函数
    return chunk.apply(some_function)

def parallel_apply(df, func, n_cores=4):
    # 将数据分成多个块
    chunks = np.array_split(df, n_cores)
    
    # 创建进程池
    pool = Pool(n_cores)
    
    # 并行处理
    results = pool.map(process_chunk, chunks)
    
    # 合并结果
    return pd.concat(results)

# 使用示例
result = parallel_apply(df, some_function)

分块处理

# 分块读取大文件
chunk_size = 10000
chunks = pd.read_csv('large_file.csv', chunksize=chunk_size)

# 分块处理
results = []
for chunk in chunks:
    # 处理每个数据块
    processed_chunk = process_chunk(chunk)
    results.append(processed_chunk)

# 合并结果
final_result = pd.concat(results)

# 使用迭代器处理大文件
def process_large_file(file_path, chunk_size=10000):
    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        # 处理每个数据块
        yield process_chunk(chunk)

大数据处理

分块读取

# 分块读取 CSV 文件
def read_csv_in_chunks(file_path, chunk_size=10000):
    return pd.read_csv(file_path, chunksize=chunk_size)

# 分块读取 Excel 文件
def read_excel_in_chunks(file_path, sheet_name=0, chunk_size=10000):
    return pd.read_excel(file_path, sheet_name=sheet_name, chunksize=chunk_size)

# 分块读取 SQL 查询结果
def read_sql_in_chunks(query, connection, chunk_size=10000):
    return pd.read_sql(query, connection, chunksize=chunk_size)

增量处理

# 增量处理数据
def incremental_processing(df, window_size=1000):
    results = []
    for i in range(0, len(df), window_size):
        chunk = df.iloc[i:i+window_size]
        # 处理数据块
        processed_chunk = process_chunk(chunk)
        results.append(processed_chunk)
    return pd.concat(results)

# 增量更新
def incremental_update(df, new_data, key_column):
    # 合并新数据
    df = pd.concat([df, new_data])
    # 删除重复项
    df = df.drop_duplicates(subset=[key_column], keep='last')
    return df

分布式处理

# 使用 Dask 进行分布式处理
import dask.dataframe as dd

# 创建 Dask DataFrame
ddf = dd.from_pandas(df, npartitions=4)

# 分布式计算
result = ddf.groupby('column').mean().compute()

# 使用 PySpark 进行分布式处理
from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder.getOrCreate()

# 将 Pandas DataFrame 转换为 Spark DataFrame
spark_df = spark.createDataFrame(df)

# 分布式计算
result = spark_df.groupBy('column').mean()

实际应用示例

示例1：大数据集处理优化

# 创建示例数据
import numpy as np
import pandas as pd

# 生成大数据集
n_rows = 1000000
df = pd.DataFrame({
    'id': range(n_rows),
    'value1': np.random.randn(n_rows),
    'value2': np.random.randn(n_rows),
    'category': np.random.choice(['A', 'B', 'C', 'D'], n_rows)
})

# 优化数据类型
df['id'] = df['id'].astype('int32')
df['value1'] = df['value1'].astype('float32')
df['value2'] = df['value2'].astype('float32')
df['category'] = df['category'].astype('category')

# 分块处理
def process_chunk(chunk):
    # 计算统计量
    stats = chunk.groupby('category').agg({
        'value1': ['mean', 'std'],
        'value2': ['mean', 'std']
    })
    return stats

# 使用分块处理
chunk_size = 100000
chunks = [df[i:i+chunk_size] for i in range(0, len(df), chunk_size)]
results = [process_chunk(chunk) for chunk in chunks]
final_result = pd.concat(results)

示例2：内存优化实践

# 创建示例数据
df = pd.DataFrame({
    'id': range(1000000),
    'float_col': np.random.randn(1000000),
    'int_col': np.random.randint(0, 100, 1000000),
    'category_col': np.random.choice(['A', 'B', 'C', 'D'], 1000000),
    'date_col': pd.date_range('2023-01-01', periods=1000000)
})

# 内存使用分析
print("优化前内存使用：")
print(df.memory_usage(deep=True).sum() / 1024**2, "MB")

# 优化数据类型
df['id'] = df['id'].astype('int32')
df['float_col'] = df['float_col'].astype('float32')
df['int_col'] = df['int_col'].astype('int16')
df['category_col'] = df['category_col'].astype('category')

# 优化后的内存使用
print("优化后内存使用：")
print(df.memory_usage(deep=True).sum() / 1024**2, "MB")

总结

性能优化部分涵盖了：

1. 内存优化（数据类型优化、内存使用分析、内存清理）
2. 计算优化（向量化操作、并行计算、分块处理）
3. 大数据处理（分块读取、增量处理、分布式处理）
4. 实际应用示例

掌握性能优化技术对于处理大规模数据至关重要，它可以帮助我们：

• 减少内存使用
• 提高计算效率
• 处理大规模数据
• 优化代码性能

建议在实际项目中注意：

• 选择合适的数据类型
• 使用向量化操作
• 合理使用分块处理
• 考虑使用分布式计算
• 定期进行性能分析
• 及时清理内存
• 优化代码结构