Python案例如何用Pandas做数据分组压测

wen python案例 2026-07-14 2

本文目录导读：

基础数据准备
基础性能测试函数
分组操作性能测试
大规模数据压测
分组数量影响测试
优化方法对比
完整压测报告
可视化结果

我来详细介绍如何使用Pandas进行数据分组压测的完整案例。

基础数据准备

import pandas as pd
import numpy as np
import time
from datetime import datetime
import random
# 创建测试数据
def create_test_data(n_rows=100000):
    """创建测试数据集"""
    np.random.seed(42)
    data = {
        'group_id': np.random.choice(['A', 'B', 'C', 'D', 'E'], n_rows),
        'category': np.random.choice(['X', 'Y', 'Z'], n_rows),
        'value1': np.random.randn(n_rows) * 100,
        'value2': np.random.randint(1, 1000, n_rows),
        'value3': np.random.uniform(0, 1, n_rows),
        'timestamp': [datetime.now() - pd.Timedelta(minutes=random.randint(0, 10000)) 
                     for _ in range(n_rows)]
    }
    return pd.DataFrame(data)
# 创建100万行测试数据
df = create_test_data(1000000)
print(f"数据集大小: {df.shape}")
print(df.head())

基础性能测试函数

def performance_test(func, df, iterations=5):
    """性能测试装饰器"""
    times = []
    for i in range(iterations):
        start_time = time.time()
        result = func(df)
        end_time = time.time()
        times.append(end_time - start_time)
    return {
        'function': func.__name__,
        'avg_time': np.mean(times),
        'min_time': np.min(times),
        'max_time': np.max(times),
        'std_time': np.std(times)
    }
def run_tests(df, test_functions):
    """运行多个测试"""
    results = []
    for func in test_functions:
        result = performance_test(func, df)
        results.append(result)
        print(f"{result['function']}: {result['avg_time']:.4f}s")
    return pd.DataFrame(results)

分组操作性能测试

# 测试不同的分组方式
def test_basic_groupby(df):
    """基础分组聚合"""
    return df.groupby('group_id')['value1'].mean()
def test_multi_column_groupby(df):
    """多列分组聚合"""
    return df.groupby(['group_id', 'category']).agg({
        'value1': 'mean',
        'value2': 'sum',
        'value3': 'std'
    })
def test_multiple_aggregations(df):
    """多种聚合操作"""
    return df.groupby('group_id').agg({
        'value1': ['mean', 'std', 'min', 'max'],
        'value2': ['sum', 'mean'],
        'value3': ['mean', 'count']
    })
def test_transform(df):
    """transform操作"""
    return df.groupby('group_id')['value1'].transform('mean')
def test_filter(df):
    """filter操作"""
    return df.groupby('group_id').filter(lambda x: x['value1'].mean() > 0)
# 运行测试
test_functions = [
    test_basic_groupby,
    test_multi_column_groupby,
    test_multiple_aggregations,
    test_transform,
    test_filter
]
results = run_tests(df, test_functions)

大规模数据压测

def stress_test_different_sizes():
    """不同数据规模的压测"""
    sizes = [10000, 50000, 100000, 500000, 1000000]
    results = []
    for size in sizes:
        df = create_test_data(size)
        # 测试不同操作
        start = time.time()
        df.groupby('group_id')['value1'].mean()
        time1 = time.time() - start
        start = time.time()
        df.groupby(['group_id', 'category']).agg({
            'value1': ['mean', 'std'],
            'value2': 'sum'
        })
        time2 = time.time() - start
        start = time.time()
        df.groupby('group_id')['value1'].transform('mean')
        time3 = time.time() - start
        results.append({
            'size': size,
            'simple_groupby': time1,
            'complex_groupby': time2,
            'transform': time3
        })
    return pd.DataFrame(results)
# 运行压测
stress_results = stress_test_different_sizes()
print("不同数据规模的压测结果:")
print(stress_results)

分组数量影响测试

def test_group_count_impact():
    """测试分组数量对性能的影响"""
    n_rows = 200000
    group_counts = [5, 10, 50, 100, 500, 1000, 5000]
    results = []
    for n_groups in group_counts:
        # 创建不同分组数量的数据
        groups = [f'Group_{i}' for i in range(n_groups)]
        data = {
            'group': np.random.choice(groups, n_rows),
            'value': np.random.randn(n_rows)
        }
        df = pd.DataFrame(data)
        # 测试性能
        start = time.time()
        for _ in range(10):
            df.groupby('group')['value'].agg(['mean', 'std', 'count'])
        total_time = time.time() - start
        results.append({
            'n_groups': n_groups,
            'avg_time': total_time / 10,
            'groups_per_row': n_rows / n_groups
        })
    return pd.DataFrame(results)
# 运行测试
group_impact = test_group_count_impact()
print("分组数量对性能的影响:")
print(group_impact)

优化方法对比

def optimization_comparison():
    """对比不同优化方法"""
    # 创建大数据集
    n_rows = 500000
    df = create_test_data(n_rows)
    results = []
    # 方法1: 基础groupby
    start = time.time()
    result1 = df.groupby('group_id')['value1'].mean()
    results.append({'method': 'basic', 'time': time.time() - start})
    # 方法2: 使用agg
    start = time.time()
    result2 = df.groupby('group_id').agg({'value1': 'mean'})
    results.append({'method': 'agg', 'time': time.time() - start})
    # 方法3: 指定索引后groupby
    start = time.time()
    df_indexed = df.set_index('group_id')
    result3 = df_indexed.groupby(level=0)['value1'].mean()
    results.append({'method': 'set_index', 'time': time.time() - start})
    # 方法4: 使用categorical类型
    start = time.time()
    df_cat = df.copy()
    df_cat['group_id'] = df_cat['group_id'].astype('category')
    result4 = df_cat.groupby('group_id')['value1'].mean()
    results.append({'method': 'categorical', 'time': time.time() - start})
    # 方法5: 先排序再groupby
    start = time.time()
    df_sorted = df.sort_values('group_id')
    result5 = df_sorted.groupby('group_id')['value1'].mean()
    results.append({'method': 'sorted', 'time': time.time() - start})
    return pd.DataFrame(results)
# 运行优化对比
opt_results = optimization_comparison()
print("\n优化方法对比:")
print(opt_results)

完整压测报告

def full_stress_test():
    """完整的压测报告"""
    print("=" * 60)
    print("Pandas GroupBy 压测报告")
    print("=" * 60)
    # 1. 基础性能测试
    print("\n1. 基础分组性能测试")
    print("-" * 40)
    df = create_test_data(100000)
    tests = {
        '简单分组': lambda d: d.groupby('group_id')['value1'].mean(),
        '多列分组': lambda d: d.groupby(['group_id', 'category']).agg({
            'value1': 'mean', 'value2': 'sum'
        }),
        '多重聚合': lambda d: d.groupby('group_id').agg({
            'value1': ['mean', 'std', 'min', 'max']
        }),
        'transform操作': lambda d: d.groupby('group_id')['value1'].transform('mean'),
        'filter操作': lambda d: d.groupby('group_id').filter(
            lambda x: x['value1'].mean() > 0
        )
    }
    for name, func in tests.items():
        times = []
        for _ in range(5):
            start = time.time()
            func(df)
            times.append(time.time() - start)
        print(f"{name}: {np.mean(times):.4f}s ({np.std(times):.4f}s)"
              f"[{np.min(times):.4f}s - {np.max(times):.4f}s]")
    # 2. 数据规模影响
    print("\n2. 数据规模影响测试")
    print("-" * 40)
    for size in [10000, 50000, 100000, 500000]:
        test_df = create_test_data(size)
        start = time.time()
        test_df.groupby('group_id')['value1'].agg(['mean', 'std'])
        elapsed = time.time() - start
        print(f"数据量 {size:>6d}: {elapsed:.4f}s")
    # 3. 内存使用情况
    print("\n3. 内存使用分析")
    print("-" * 40)
    print(f"DataFrame内存使用: {df.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB")
    # 4. 建议
    print("\n4. 优化建议")
    print("-" * 40)
    print("""
    1. 使用categorical类型优化分组列
    2. 避免过多的聚合操作
    3. 考虑使用transform代替apply
    4. 大数据集时考虑采样测试
    5. 使用合适的索引策略
    """)
# 运行完整压测
full_stress_test()

可视化结果

import matplotlib.pyplot as plt
def visualize_results():
    """可视化压测结果"""
    fig, axes = plt.subplots(2, 2, figsize=(12, 10))
    # 数据规模性能图
    sizes = [10000, 50000, 100000, 500000, 1000000]
    times = []
    for size in sizes:
        df = create_test_data(size)
        start = time.time()
        df.groupby('group_id')['value1'].mean()
        times.append(time.time() - start)
    axes[0, 0].plot(sizes, times, 'b-o')
    axes[0, 0].set_title('数据规模vs执行时间')
    axes[0, 0].set_xlabel('数据量')
    axes[0, 0].set_ylabel('执行时间(s)')
    # 分组数量影响
    group_counts = [5, 10, 50, 100, 500, 1000]
    group_times = []
    for n_groups in group_counts:
        groups = [f'G{i}' for i in range(n_groups)]
        df = pd.DataFrame({
            'group': np.random.choice(groups, 100000),
            'value': np.random.randn(100000)
        })
        start = time.time()
        df.groupby('group')['value'].mean()
        group_times.append(time.time() - start)
    axes[0, 1].plot(group_counts, group_times, 'r-s')
    axes[0, 1].set_title('分组数量vs执行时间')
    axes[0, 1].set_xlabel('分组数')
    axes[0, 1].set_ylabel('执行时间(s)')
    plt.tight_layout()
    plt.show()
# visualize_results()  # 取消注释以显示图表

这个压测案例涵盖了：

基础性能测试：不同分组操作的执行时间
大规模数据测试：不同数据量下的性能表现
分组数量影响：分组数量对性能的影响
优化方法对比：不同优化策略的效果
完整压测报告：综合性的性能分析

使用时可以根据实际需求调整测试参数和数据量。