Python案例如何用Pandas做数据分组重采高可用

wen python案例 2026-07-14 272

本文目录导读：

基础环境准备
数据分组操作
时间序列重采样
分组+重采样（高可用方案）
实际应用案例
性能优化技巧
异常处理和监控

我来为你详细介绍如何使用Pandas进行数据分组和高可用重采样,我会通过多个实用案例来展示。

基础环境准备

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 创建示例数据
np.random.seed(42)
# 生成时间序列数据
dates = pd.date_range('2024-01-01', periods=1000, freq='H')
data = pd.DataFrame({
    'date': dates,
    'group': np.random.choice(['A', 'B', 'C'], 1000),
    'value1': np.random.randn(1000) * 10 + 50,
    'value2': np.random.randn(1000) * 5 + 20
})
print("数据预览:")
print(data.head())
print(f"数据形状: {data.shape}")

数据分组操作

1 基本分组统计

# 按组进行基础统计
group_stats = data.groupby('group').agg({
    'value1': ['mean', 'std', 'min', 'max'],
    'value2': ['mean', 'std', 'min', 'max']
})
print("分组统计结果:")
print(group_stats)
# 分组并计算多个统计量
detailed_stats = data.groupby('group').agg(
    value1_mean=('value1', 'mean'),
    value1_median=('value1', 'median'),
    value1_std=('value1', 'std'),
    value2_mean=('value2', 'mean'),
    count=('value1', 'count')
)
print("\n详细分组统计:")
print(detailed_stats)

2 多级分组

# 创建更多分组维度
data['category'] = np.random.choice(['X', 'Y', 'Z'], 1000)
data['month'] = data['date'].dt.month
# 多级分组
multi_group = data.groupby(['group', 'category', 'month']).agg({
    'value1': ['mean', 'count'],
    'value2': 'sum'
})
print("多级分组结果:")
print(multi_group.head(12))
# 重置索引便于查看
multi_group_flat = multi_group.reset_index()
print("\n重置索引后的数据:")
print(multi_group_flat.head())

时间序列重采样

1 设置时间索引

# 将date列设置为索引
data_ts = data.set_index('date')
print("时间序列数据:")
print(data_ts.head())

2 不同频率的重采样

# 日重采样
daily_resample = data_ts.resample('D').agg({
    'value1': ['mean', 'max', 'min'],
    'value2': 'sum'
})
print("日重采样结果:")
print(daily_resample.head())
# 周重采样
weekly_resample = data_ts.resample('W').agg({
    'value1': 'mean',
    'value2': 'sum'
})
print("\n周重采样结果:")
print(weekly_resample.head())
# 月重采样
monthly_resample = data_ts.resample('M').agg({
    'value1': ['mean', 'std'],
    'value2': ['sum', 'mean']
})
print("\n月重采样结果:")
print(monthly_resample.head())

分组+重采样（高可用方案）

1 按组重采样

# 按组分组的重采样
def resample_by_group(df, group_col='group', date_col='date', freq='D'):
    """
    按组分组的重采样函数
    Parameters:
    -----------
    df : DataFrame
    group_col : str, 分组列名
    date_col : str, 日期列名
    freq : str, 重采样频率
    Returns:
    --------
    DataFrame : 重采样后的数据
    """
    # 确保日期列是datetime类型
    df[date_col] = pd.to_datetime(df[date_col])
    # 按组设置索引并重采样
    result = df.set_index(date_col).groupby(group_col).resample(freq).mean()
    return result
# 使用函数进行分组重采样
group_resampled = resample_by_group(data, freq='W')
print("分组周重采样结果:")
print(group_resampled.head(12))

2 高性能分组重采样

import pandas as pd
from concurrent.futures import ThreadPoolExecutor
import functools
class HighAvailabilityResampler:
    """高可用分组重采样器"""
    def __init__(self, df, date_col='date', group_col='group'):
        self.df = df
        self.date_col = date_col
        self.group_col = group_col
        self.df[date_col] = pd.to_datetime(self.df[date_col])
    def resample_group(self, group_data, freq='D', agg_func='mean'):
        """对单个组进行重采样"""
        try:
            result = (group_data
                     .set_index(self.date_col)
                     .resample(freq)
                     .agg(agg_func))
            result[self.group_col] = group_data[self.group_col].iloc[0]
            return result
        except Exception as e:
            print(f"重采样错误: {e}")
            return None
    def parallel_resample(self, freq='D', agg_func='mean', n_workers=4):
        """并行重采样"""
        # 按组分批
        groups = [group for name, group in self.df.groupby(self.group_col)]
        # 使用线程池并行处理
        with ThreadPoolExecutor(max_workers=n_workers) as executor:
            partial_func = functools.partial(
                self.resample_group, 
                freq=freq, 
                agg_func=agg_func
            )
            results = list(executor.map(partial_func, groups))
        # 合并结果
        final_result = pd.concat(results, ignore_index=False)
        return final_result.sort_index()
# 使用高可用重采样器
resampler = HighAvailabilityResampler(data)
result_high_avail = resampler.parallel_resample(freq='D', n_workers=3)
print("高可用并行重采样结果:")
print(result_high_avail.head())

实际应用案例

1 股票数据分组重采样

# 模拟股票数据
def generate_stock_data(n_stocks=5, days=365):
    """生成模拟股票数据"""
    stocks = [f'Stock_{i}' for i in range(n_stocks)]
    dates = pd.date_range('2024-01-01', periods=days, freq='D')
    data_list = []
    for stock in stocks:
        base_price = np.random.uniform(50, 200)
        prices = np.random.randn(days) * 2 + base_price
        prices = np.abs(prices) * 0.5 + np.cumsum(np.random.randn(days) * 0.5)
        prices = base_price + prices.cumsum()
        df = pd.DataFrame({
            'date': dates,
            'stock': stock,
            'open': prices,
            'high': prices * (1 + np.random.uniform(0, 0.05, days)),
            'low': prices * (1 - np.random.uniform(0, 0.05, days)),
            'close': prices * (1 + np.random.randn(days) * 0.02),
            'volume': np.random.randint(1000, 100000, days)
        })
        data_list.append(df)
    return pd.concat(data_list, ignore_index=True)
# 生成股票数据
stock_data = generate_stock_data(5, 365)
print("股票数据预览:")
print(stock_data.head())
# 按股票分组并计算周度统计
weekly_stock_stats = (stock_data
                     .set_index('date')
                     .groupby('stock')
                     .resample('W')
                     .agg({
                         'open': 'first',
                         'high': 'max',
                         'low': 'min',
                         'close': 'last',
                         'volume': 'sum'
                     }))
print("\n股票周度统计:")
print(weekly_stock_stats.head(10))

2 传感器数据分组重采样

# 模拟物联网传感器数据
def generate_sensor_data(n_sensors=10, hours=720):
    """生成模拟传感器数据"""
    sensors = [f'Sensor_{i:03d}' for i in range(n_sensors)]
    timestamps = pd.date_range('2024-01-01', periods=hours, freq='H')
    data_list = []
    for sensor in sensors:
        # 每个传感器有不同的基准值
        base_temp = np.random.uniform(20, 30)
        base_humidity = np.random.uniform(40, 70)
        df = pd.DataFrame({
            'timestamp': timestamps,
            'sensor_id': sensor,
            'temperature': base_temp + np.random.randn(hours) * 2,
            'humidity': base_humidity + np.random.randn(hours) * 5,
            'pressure': 1013 + np.random.randn(hours) * 10,
            'battery': np.random.uniform(3.3, 4.2, hours)
        })
        data_list.append(df)
    return pd.concat(data_list, ignore_index=True)
# 生成传感器数据
sensor_data = generate_sensor_data(10, 720)
print("传感器数据预览:")
print(sensor_data.head())
# 按传感器分组进行日度统计
daily_sensor_stats = (sensor_data
                     .set_index('timestamp')
                     .groupby('sensor_id')
                     .resample('D')
                     .agg({
                         'temperature': ['mean', 'min', 'max', 'std'],
                         'humidity': ['mean', 'std'],
                         'pressure': 'mean',
                         'battery': 'min'
                     }))
print("\n传感器日度统计:")
print(daily_sensor_stats.head(10))

性能优化技巧

class OptimizedResampler:
    """优化的分组重采样器"""
    def __init__(self, df):
        self.df = df
    def chunked_resample(self, group_col='group', date_col='date', 
                        freq='D', chunksize=1000):
        """分块重采样，适用于大数据"""
        chunks = []
        for chunk_start in range(0, len(self.df), chunksize):
            chunk = self.df.iloc[chunk_start:chunk_start + chunksize]
            # 处理当前块
            processed = (chunk
                        .set_index(date_col)
                        .groupby(group_col)
                        .resample(freq)
                        .mean()
                        .fillna(method='ffill'))
            chunks.append(processed)
        return pd.concat(chunks)
    def cached_resample(self, group_col='group', date_col='date', freq='D'):
        """带缓存的重采样"""
        from functools import lru_cache
        @lru_cache(maxsize=128)
        def _cached_resample(group_name, start_date, end_date, freq):
            mask = (self.df[group_col] == group_name) & \
                   (self.df[date_col] >= start_date) & \
                   (self.df[date_col] <= end_date)
            subset = self.df[mask]
            return (subset
                   .set_index(date_col)
                   .resample(freq)
                   .mean())
        # 使用缓存的示例
        results = []
        for group in self.df[group_col].unique():
            date_range = self.df[self.df[group_col] == group][date_col]
            result = _cached_resample(
                group, 
                date_range.min(), 
                date_range.max(), 
                freq
            )
            results.append(result)
        return pd.concat(results)
# 性能测试
import time
# 创建较大数据集
large_data = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=10000, freq='H'),
    'group': np.random.choice(['A', 'B', 'C', 'D', 'E'], 10000),
    'value1': np.random.randn(10000),
    'value2': np.random.randn(10000)
})
# 测试不同方法
print("性能测试结果:")
# 方法1: 直接重采样
start = time.time()
result1 = (large_data
          .set_index('date')
          .groupby('group')
          .resample('D')
          .mean())
print(f"直接重采样: {time.time()-start:.3f}秒")
# 方法2: 使用优化重采样器
start = time.time()
optimized = OptimizedResampler(large_data)
result2 = optimized.chunked_resample(chunksize=2000)
print(f"分块重采样: {time.time()-start:.3f}秒")

异常处理和监控

import logging
from datetime import datetime
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RobustResampler:
    """健壮的分组重采样器，带有异常处理和监控"""
    def __init__(self, df):
        self.df = df
        self.metrics = {
            'total_groups': 0,
            'successful_groups': 0,
            'failed_groups': 0,
            'total_records': len(df)
        }
    def resample_with_fallback(self, group_col='group', date_col='date', 
                              freq='D', agg_func='mean'):
        """带失败转移的重采样"""
        result_data = []
        for name, group in self.df.groupby(group_col):
            self.metrics['total_groups'] += 1
            try:
                # 尝试标准重采样
                resampled = (group
                           .set_index(date_col)
                           .resample(freq)
                           .agg(agg_func))
                self.metrics['successful_groups'] += 1
                result_data.append(resampled)
            except Exception as e:
                self.metrics['failed_groups'] += 1
                logger.warning(f"组 {name} 重采样失败: {e}")
                # 尝试替代方法
                try:
                    # 使用前向填充作为回退
                    resampled = (group
                               .set_index(date_col)
                               .resample(freq)
                               .ffill()
                               .iloc[0:1])
                    logger.info(f"组 {name} 使用前向填充成功")
                    self.metrics['successful_groups'] += 1
                    result_data.append(resampled)
                except:
                    logger.error(f"组 {name} 所有方法都失败")
        # 记录统计信息
        self._log_metrics()
        return pd.concat(result_data, ignore_index=False) if result_data else pd.DataFrame()
    def _log_metrics(self):
        """记录处理指标"""
        logger.info(f"""
        处理完成:
        - 总组数: {self.metrics['total_groups']}
        - 成功: {self.metrics['successful_groups']}
        - 失败: {self.metrics['failed_groups']}
        - 总记录数: {self.metrics['total_records']}
        """)
# 使用鲁棒重采样器
robust_resampler = RobustResampler(large_data)
robust_result = robust_resampler.resample_with_fallback(freq='W', agg_func='mean')
print("\n鲁棒重采样结果:")
print(robust_result.head())

案例展示了Pandas在数据分组和重采样方面的强大功能：