如何快速对百万行多列DataFrame聚合多种统计指标？

刚好处理过类似的超大型DataFrame聚合需求，给你几个能显著提升速度的方案，针对你要计算的mean、sd、3rd/4th矩以及Sarle's系数这些指标：

1. 优化Pandas聚合：内置函数+自定义矢量化函数结合 #

首先修正你原代码的小问题：不能直接把列索引传入groupby的参数，应该先指定目标列，再通过agg()批量绑定多个统计指标。同时尽量用矢量化函数代替循环，减少开销。

代码示例： #

import pandas as pd
from scipy.stats import moment
import numpy as np

# 定义目标列（前100列）
target_cols = df.columns[:100]

# 自定义Sarle's系数函数，处理标准差为0的边界情况
def sarle_coeff(x):
    mean_val = x.mean()
    median_val = x.median()
    std_val = x.std(ddof=1)  # 用样本标准差
    return (mean_val - median_val) / std_val if std_val != 0 else 0

# 构建聚合字典：每个列对应需要计算的所有指标
agg_spec = {
    col: [
        'mean',
        'std',  # 对应你要的sd
        lambda x: moment(x, moment=3),
        lambda x: moment(x, moment=4),
        sarle_coeff
    ] for col in target_cols
}

# 执行分组聚合，as_index=False让id保留为普通列
agg_result = df.groupby('id', as_index=False)[target_cols].agg(agg_spec)

# 可选：重命名列，让结果更清晰
agg_result.columns = ['_'.join(col).strip() for col in agg_result.columns.values]

2. 用Numpy矢量化批量计算（速度更快） #

如果Pandas的agg()还是不够快，直接用Numpy的矢量化操作处理分组数据会更高效——因为Numpy底层是C实现，避免了Pandas的一些额外开销。适合列数多、数据量极大的场景。

代码示例： #

import pandas as pd
import numpy as np

target_cols = df.columns[:100]
groups = df.groupby('id')[target_cols]

# 初始化结果存储字典
result_data = {'id': []}
# 为每个列的每个指标创建存储键
for col in target_cols:
    result_data[f'{col}_mean'] = []
    result_data[f'{col}_std'] = []
    result_data[f'{col}_moment3'] = []
    result_data[f'{col}_moment4'] = []
    result_data[f'{col}_sarle'] = []

# 遍历每个分组，批量计算所有列的统计量
for id_val, group_df in groups:
    result_data['id'].append(id_val)
    # 转成Numpy数组，方便批量操作
    group_arr = group_df.to_numpy()
    
    # 一次性计算所有列的基础统计量
    means = np.mean(group_arr, axis=0)
    stds = np.std(group_arr, axis=0, ddof=1)
    medians = np.median(group_arr, axis=0)
    # 计算中心矩：先中心化数据
    centered_arr = group_arr - means
    moment3 = np.mean(centered_arr ** 3, axis=0)
    moment4 = np.mean(centered_arr ** 4, axis=0)
    # 计算Sarle's系数，处理std为0的情况
    sarle_vals = (means - medians) / stds
    sarle_vals[stds == 0] = 0  # 避免除以0错误
    
    # 将结果填充到字典中
    for idx, col in enumerate(target_cols):
        result_data[f'{col}_mean'].append(means[idx])
        result_data[f'{col}_std'].append(stds[idx])
        result_data[f'{col}_moment3'].append(moment3[idx])
        result_data[f'{col}_moment4'].append(moment4[idx])
        result_data[f'{col}_sarle'].append(sarle_vals[idx])

# 转成最终的DataFrame
agg_result = pd.DataFrame(result_data)

3. 内存不足时用Dask分块并行计算 #

如果你的数据大到内存放不下（比如数百万行+数百列），可以用Dask DataFrame进行分块处理，自动并行计算，充分利用CPU多核资源。

代码示例： #

import dask.dataframe as dd
from scipy.stats import moment

target_cols = df.columns[:100]

# 自定义Sarle's系数函数（和之前一致）
def sarle_coeff(x):
    mean_val = x.mean()
    median_val = x.median()
    std_val = x.std(ddof=1)
    return (mean_val - median_val) / std_val if std_val != 0 else 0

# 将Pandas DataFrame转为Dask DataFrame，分块数根据CPU核心数调整
ddf = dd.from_pandas(df, npartitions=4)

# 构建聚合规则
agg_spec = {
    col: [
        'mean',
        'std',
        lambda x: moment(x, moment=3),
        lambda x: moment(x, moment=4),
        sarle_coeff
    ] for col in target_cols
}

# 执行聚合，compute()触发实际计算
agg_result = ddf.groupby('id')[target_cols].agg(agg_spec).compute()

选择建议 #

• 如果数据能放进内存，优先用Numpy矢量化方案，速度比Pandas agg快2-5倍；
• 如果需要代码简洁，用优化后的Pandas agg方案；
• 如果内存不够，直接上Dask分块计算，无需修改太多代码就能处理超大型数据。

内容的提问来源于stack exchange，提问作者roger