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

wen python案例 2026-07-14 7

本文目录导读：

基础数据准备
不同重采样方法性能对比
压力测试框架
优化策略测试
内存使用监控
可视化结果
优化建议总结

我来详细说明如何使用Pandas进行数据分组重采样和压力测试。

基础数据准备

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import time
# 创建测试数据
def create_test_data(n_rows=10000):
    """生成测试数据集"""
    np.random.seed(42)
    # 生成时间序列
    base_time = datetime(2023, 1, 1)
    timestamps = [base_time + timedelta(minutes=i) for i in range(n_rows)]
    # 创建多组数据
    groups = ['A', 'B', 'C', 'D']
    data = {
        'timestamp': timestamps * (n_rows // len(timestamps)) + timestamps[:n_rows % len(timestamps)],
        'group': np.random.choice(groups, n_rows),
        'value1': np.random.randn(n_rows) * 100,
        'value2': np.random.randn(n_rows) * 50,
        'category': np.random.choice(['X', 'Y', 'Z'], n_rows)
    }
    df = pd.DataFrame(data)
    # 只保留前n_rows行
    df = df.head(n_rows)
    return df
# 创建100万行测试数据
df_large = create_test_data(1000000)
print(f"数据集大小: {df_large.shape}")

不同重采样方法性能对比

def benchmark_resampling_methods(df):
    """对比不同重采样方法的性能"""
    results = []
    # 方法1: 使用resample和groupby
    def method_1(df):
        return (df.set_index('timestamp')
                .groupby('group')
                .resample('1H')
                .agg({'value1': 'mean', 'value2': 'sum'})
                .reset_index())
    # 方法2: 先分组再重采样
    def method_2(df):
        def resample_group(group):
            return (group.set_index('timestamp')
                   .resample('1H')
                   .agg({'value1': 'mean', 'value2': 'sum'}))
        return df.groupby('group').apply(resample_group).reset_index()
    # 方法3: 使用Grouper
    def method_3(df):
        return (df.set_index('timestamp')
                .groupby(['group', pd.Grouper(freq='1H')])
                .agg({'value1': 'mean', 'value2': 'sum'})
                .reset_index())
    # 方法4: 先排序再重采样
    def method_4(df):
        df_sorted = df.sort_values(['group', 'timestamp'])
        return (df_sorted.set_index('timestamp')
                .groupby('group')
                .resample('1H')
                .agg({'value1': 'mean', 'value2': 'sum'})
                .reset_index())
    methods = [
        ('Method 1: GroupBy + Resample', method_1),
        ('Method 2: GroupBy + Apply', method_2),
        ('Method 3: Grouper', method_3),
        ('Method 4: Sort + Resample', method_4)
    ]
    for name, method in methods:
        # 使用小样本测试
        df_sample = df.head(100000)
        start_time = time.time()
        try:
            result = method(df_sample)
            elapsed = time.time() - start_time
            results.append({
                'method': name,
                'time': elapsed,
                'rows': len(result),
                'success': True
            })
        except Exception as e:
            results.append({
                'method': name,
                'time': None,
                'rows': 0,
                'success': False,
                'error': str(e)
            })
        print(f"{name}: {elapsed:.3f}s")
    return pd.DataFrame(results)
# 运行性能测试
perf_results = benchmark_resampling_methods(df_large)
print("\n性能测试结果:")
print(perf_results)

压力测试框架

class ResampleStressTest:
    """重采样压力测试类"""
    def __init__(self):
        self.results = []
    def test_different_sizes(self, sizes=[100000, 500000, 1000000]):
        """测试不同数据量"""
        for size in sizes:
            print(f"\n=== 测试数据量: {size:,} 行 ===")
            df = create_test_data(size)
            start_time = time.time()
            result = self._perform_resampling(df)
            elapsed = time.time() - start_time
            self.results.append({
                'data_size': size,
                'time_seconds': elapsed,
                'rows_per_second': size / elapsed,
                'result_rows': len(result)
            })
            print(f"耗时: {elapsed:.3f}s")
            print(f"处理速度: {size/elapsed:,.0f} 行/秒")
    def test_different_frequencies(self, df, frequencies=['10T', '30T', '1H', '6H', '1D']):
        """测试不同重采样频率"""
        print("\n=== 测试不同重采样频率 ===")
        for freq in frequencies:
            start_time = time.time()
            result = self._perform_resampling(df, freq)
            elapsed = time.time() - start_time
            self.results.append({
                'frequency': freq,
                'time_seconds': elapsed,
                'result_rows': len(result)
            })
            print(f"频率 {freq}: {elapsed:.3f}s, 结果行数: {len(result):,}")
    def test_different_aggregations(self, df, agg_functions=['mean', 'sum', 'count', 'std', 'min', 'max']):
        """测试不同聚合函数"""
        print("\n=== 测试不同聚合函数 ===")
        for agg_func in agg_functions:
            start_time = time.time()
            result = self._perform_resampling_with_agg(df, agg_func)
            elapsed = time.time() - start_time
            self.results.append({
                'aggregation': agg_func,
                'time_seconds': elapsed,
                'result_rows': len(result)
            })
            print(f"聚合函数 {agg_func}: {elapsed:.3f}s")
    def _perform_resampling(self, df, freq='1H'):
        """执行重采样"""
        return (df.set_index('timestamp')
                .groupby('group')
                .resample(freq)
                .agg({'value1': ['mean', 'std'], 'value2': ['sum', 'count']})
                .reset_index())
    def _perform_resampling_with_agg(self, df, agg_func):
        """执行带特定聚合的重采样"""
        return (df.set_index('timestamp')
                .groupby('group')
                .resample('1H')
                .agg({'value1': agg_func, 'value2': agg_func})
                .reset_index())
    def run_comprehensive_test(self):
        """运行综合测试"""
        print("=" * 50)
        print("重采样压力测试 - 综合测试")
        print("=" * 50)
        # 测试不同数据量
        self.test_different_sizes([50000, 100000, 200000])
        # 测试不同频率
        df = create_test_data(200000)
        self.test_different_frequencies(df)
        # 测试不同聚合函数
        self.test_different_aggregations(df)
        return pd.DataFrame(self.results)
# 运行压力测试
tester = ResampleStressTest()
results_df = tester.run_comprehensive_test()

优化策略测试

def optimization_strategies():
    """测试不同优化策略"""
    print("\n=== 优化策略测试 ===")
    df = create_test_data(500000)
    strategies = {
        'Base': lambda df: (df.set_index('timestamp')
                           .groupby('group')
                           .resample('1H')
                           .agg({'value1': 'mean', 'value2': 'sum'})),
        'Sorted': lambda df: (df.sort_values(['group', 'timestamp'])
                             .set_index('timestamp')
                             .groupby('group')
                             .resample('1H')
                             .agg({'value1': 'mean', 'value2': 'sum'})),
        'Pivot': lambda df: (df.pivot_table(
                            index='group', 
                            columns=pd.Grouper(key='timestamp', freq='1H'),
                            values='value1',
                            aggfunc='mean')),
        'Dask': lambda df: (df.set_index('timestamp')
                           .groupby('group')
                           .resample('1H')
                           .agg('mean')
                           .dropna())
    }
    results = []
    for name, func in strategies.items():
        start_time = time.time()
        try:
            result = func(df)
            elapsed = time.time() - start_time
            results.append({'strategy': name, 'time': elapsed, 'status': 'OK'})
        except Exception as e:
            elapsed = time.time() - start_time
            results.append({'strategy': name, 'time': elapsed, 'status': f'Error: {str(e)}'})
        print(f"{name}: {elapsed:.3f}s")
    return pd.DataFrame(results)
# 运行优化测试
opt_results = optimization_strategies()

内存使用监控

import psutil
import os
def monitor_memory_usage():
    """监控内存使用情况"""
    process = psutil.Process(os.getpid())
    print("\n=== 内存使用监控 ===")
    # 创建大数据集
    df = create_test_data(1000000)
    initial_memory = process.memory_info().rss / 1024 / 1024
    print(f"初始内存使用: {initial_memory:.2f} MB")
    # 执行重采样并监控内存
    memory_log = []
    for i in range(5):
        start_time = time.time()
        # 执行重采样
        result = (df.set_index('timestamp')
                  .groupby('group')
                  .resample('1H')
                  .agg('mean'))
        elapsed = time.time() - start_time
        current_memory = process.memory_info().rss / 1024 / 1024
        memory_log.append({
            'iteration': i + 1,
            'time_seconds': elapsed,
            'memory_mb': current_memory,
            'memory_increase': current_memory - initial_memory
        })
        print(f"迭代 {i+1}: 耗时={elapsed:.3f}s, 内存={current_memory:.2f}MB")
    return pd.DataFrame(memory_log)
# 运行内存监控
memory_df = monitor_memory_usage()

可视化结果

import matplotlib.pyplot as plt
import seaborn as sns
def plot_stress_test_results(results_df):
    """可视化压力测试结果"""
    # 创建图表
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    # 1. 数据量 vs 耗时
    if 'data_size' in results_df.columns:
        ax1 = axes[0, 0]
        ax1.plot(results_df['data_size'], results_df['time_seconds'], 'bo-')
        ax1.set_xlabel('数据量 (行)')
        ax1.set_ylabel('耗时 (秒)')
        ax1.set_title('数据量 vs 耗时')
        ax1.grid(True)
    # 2. 处理速度
    if 'rows_per_second' in results_df.columns:
        ax2 = axes[0, 1]
        ax2.bar(range(len(results_df)), results_df['rows_per_second'])
        ax2.set_xlabel('测试序号')
        ax2.set_ylabel('处理速度 (行/秒)')
        ax2.set_title('处理速度对比')
    # 3. 频率 vs 结果行数
    if 'frequency' in results_df.columns:
        ax3 = axes[1, 0]
        ax3.bar(results_df['frequency'], results_df['result_rows'])
        ax3.set_xlabel('重采样频率')
        ax3.set_ylabel('结果行数')
        ax3.set_title('重采样频率 vs 结果行数')
    # 4. 聚合函数性能
    if 'aggregation' in results_df.columns:
        ax4 = axes[1, 1]
        ax4.bar(results_df['aggregation'], results_df['time_seconds'])
        ax4.set_xlabel('聚合函数')
        ax4.set_ylabel('耗时 (秒)')
        ax4.set_title('聚合函数性能对比')
        plt.setp(ax4.xaxis.get_majorticklabels(), rotation=45)
    plt.tight_layout()
    plt.show()
# 绘制压力测试结果
plot_stress_test_results(results_df)

优化建议总结

def generate_optimization_tips():
    """生成优化建议"""
    tips = """
=== Pandas分组重采样优化建议 ===
1. **数据预处理优化**
   - 确保时间索引唯一且排序
   - 使用category类型处理分组列
   - 提前过滤不需要的数据
2. **索引优化**
   - 优先使用set_index('timestamp')设置时间索引
   - 使用sort_index()对时间索引排序
   - 避免在groupby后重复设置索引
3. **重采样策略**
   - 使用pd.Grouper进行分组重采样
   - 选择合适的重采样频率
   - 避免过细的重采样粒度
4. **聚合优化**
   - 使用字典指定不同列的聚合函数
   - 优先使用内置聚合函数
   - 减少链式操作
5. **内存优化**
   - 使用chunksize分批处理大数据集
   - 及时释放不用的DataFrame
   - 使用copy=False避免不必要的数据复制
6. **并行处理**
   - 使用swifter或dask进行并行重采样
   - 按group分块并行处理
   - 利用多核CPU优势
7. **最佳实践**
   - 先排序再分组重采样
   - 避免使用apply方法
   - 使用agg替代apply进行聚合
   """
    print(tips)
# 输出优化建议
generate_optimization_tips()

这个完整的案例展示了：

数据准备和基础重采样
多种重采样方法的性能对比
全面的压力测试框架
不同维度的性能测试（数据量、频率、聚合函数）
内存使用监控
结果可视化
优化建议

通过这些测试,你可以找出最适合你数据规模和业务场景的重采样策略。