当前位置：首页 > ds >正文

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

ds 2025/7/3 8:13:07

文章概要

本文详细介绍 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 Pooldef 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")