优化Pandas多重采样操作：提升传感器数据处理效率

我有一个包含动作类型、x/y/z坐标、标签及时间戳的数据集，共约5万条记录，部分采样频率达20次/秒。需按1秒、3秒、5秒的时间间隔重采样，并基于坐标计算衍生特征。当前实现脚本可完成任务，但效率极低，耗时数小时。曾尝试使用df.resample(....).last()一次性重采样，却生成了远超原数据量的2995753行数据，现寻求该脚本的优化方案。

原代码如下：

type = 16
bin = '5S'

type_str = 'accelerometer' if type == 2 else 'gyroscope'
name = str(type) + "_" + str(type_str)

# data load and basic preparation
csv = pd.read_csv('data/cleaned_' + str(type) + '.csv', sep=',')
df = pd.DataFrame(csv)
df.sort_values(by='date', inplace=True)

df['date'] = pd.to_datetime(df['date'])
df['label'] = df['label'].astype('int')
# df['time'] = df['date'].dt.time

resampled = pd.DataFrame({})  # empty df for the new features

resampled['label'] = df.resample(bin, on="date")['label'].mean()  # get the avg value for the target action / the label

resampled['n'] = df.resample(bin, on="date")['label'].count()

resampled['x'] = df.resample(bin, on="date").x.mean()
resampled['y'] = df.resample(bin, on="date").y.mean()
resampled['z'] = df.resample(bin, on="date").z.mean()

# get the max value for each dimension for every group
resampled['x_avg_max'] = df.resample(bin, on="date").x.max()
resampled['x_avg_min'] = df.resample(bin, on="date").x.min()
resampled['x_range'] = resampled['x_avg_max'] - resampled['x_avg_min']
resampled['x_standard_deviation'] = df.resample(bin, on="date").x.std()
resampled['x_variance'] = df.resample(bin, on="date").x.var()
resampled['x_skew'] = df.resample(bin, on="date").x.agg(skew)
resampled['x_kurtosis'] = df.resample(bin, on="date").x.agg(kurtosis)

resampled['y_avg_max'] = df.resample(bin, on="date").y.max()
resampled['y_avg_min'] = df.resample(bin, on="date").y.min()
resampled['y_range'] = resampled['y_avg_max'] - resampled['y_avg_min']
resampled['y_standard_deviation'] = df.resample(bin, on="date").y.std()
resampled['y_variance'] = df.resample(bin, on="date").y.var()
resampled['y_skew'] = df.resample(bin, on="date").y.agg(skew)
resampled['y_kurtosis'] = df.resample(bin, on="date").y.agg(kurtosis)

resampled['z_avg_max'] = df.resample(bin, on="date").z.max()
resampled['z_avg_min'] = df.resample(bin, on="date").z.min()
resampled['z_range'] = resampled['z_avg_max'] - resampled['z_avg_min']
resampled['z_standard_deviation'] = df.resample(bin, on="date").z.std()
resampled['z_variance'] = df.resample(bin, on="date").z.var()
resampled['z_skew'] = df.resample(bin, on="date").z.agg(skew)
resampled['z_kurtosis'] = df.resample(bin, on="date").z.agg(kurtosis)

resampled.fillna(0, inplace=True)  # replace empty values with zeros
# new features
resampled['svm'] = np.sqrt(resampled['x'] * resampled['x'] + resampled['y'] * resampled['y'] + resampled['z'] * resampled['z'])
resampled['vmm'] = resampled[["x", "y", "x"]].max(axis=1)  # get the max value for each time group, regardless if it is x, y, or z
resampled['sma'] = abs(resampled['x']) + abs(resampled['y']) + abs(resampled['z'])
resampled['ai'] = resampled['svm'] / resampled['n']

# remove rows if n=0 -> no measurements from the sensors
resampled = resampled.drop(resampled[resampled.n == 0].index)

print(resampled)
resampled.to_csv("data/resampled/resampled_" + str(type) + "_" + str(bin) + ".csv")

核心优化点 #

• 仅执行一次resample操作：原代码重复调用df.resample()数十次，每次都要重新分组计算，这是性能低下的主要原因。先创建一次resample分组对象，再基于该对象计算所有特征。
• 避免生成空时间槽：默认resample会生成时间范围内所有间隔的行，哪怕对应时间段没有数据。通过dropna过滤无数据行，减少无效计算。
• 批量统计特征计算：使用agg方法一次性为x/y/z列计算多个统计量，减少代码冗余和计算开销。
• 向量化衍生特征计算：保持Pandas的向量化操作，避免逐行处理，提升计算效率。

优化后的代码 #

import pandas as pd
import numpy as np
from scipy.stats import skew, kurtosis

type = 16
bin = '5S'

type_str = 'accelerometer' if type == 2 else 'gyroscope'
name = str(type) + "_" + str(type_str)

# 数据加载与预处理
df = pd.read_csv(f'data/cleaned_{type}.csv', sep=',')
df.sort_values(by='date', inplace=True)
df['date'] = pd.to_datetime(df['date'])
df['label'] = df['label'].astype('int')

# 关键：仅创建一次resample分组对象，避免重复计算
resampler = df.resample(bin, on='date')

# 批量定义各列需要计算的统计量
agg_spec = {
    'label': ['mean', 'count'],
    'x': ['mean', 'max', 'min', 'std', 'var', skew, kurtosis],
    'y': ['mean', 'max', 'min', 'std', 'var', skew, kurtosis],
    'z': ['mean', 'max', 'min', 'std', 'var', skew, kurtosis]
}

# 执行聚合计算并扁平化列名
resampled = resampler.agg(agg_spec)
resampled.columns = ['_'.join(col).strip() for col in resampled.columns.values]

# 重命名列以匹配原逻辑
resampled.rename(columns={
    'label_mean': 'label',
    'label_count': 'n',
    'x_mean': 'x',
    'x_max': 'x_avg_max',
    'x_min': 'x_avg_min',
    'x_std': 'x_standard_deviation',
    'x_var': 'x_variance',
    'x_skew': 'x_skew',
    'x_kurtosis': 'x_kurtosis',
    'y_mean': 'y',
    'y_max': 'y_avg_max',
    'y_min': 'y_avg_min',
    'y_std': 'y_standard_deviation',
    'y_var': 'y_variance',
    'y_skew': 'y_skew',
    'y_kurtosis': 'y_kurtosis',
    'z_mean': 'z',
    'z_max': 'z_avg_max',
    'z_min': 'z_avg_min',
    'z_std': 'z_standard_deviation',
    'z_var': 'z_variance',
    'z_skew': 'z_skew',
    'z_kurtosis': 'z_kurtosis'
}, inplace=True)

# 计算范围特征
resampled['x_range'] = resampled['x_avg_max'] - resampled['x_avg_min']
resampled['y_range'] = resampled['y_avg_max'] - resampled['y_avg_min']
resampled['z_range'] = resampled['z_avg_max'] - resampled['z_avg_min']

# 过滤无数据行，处理剩余缺失值
resampled = resampled.dropna(subset=['n'])
resampled['n'] = resampled['n'].astype(int)
resampled.fillna(0, inplace=True)

# 计算衍生特征（向量化操作，修正原代码笔误）
resampled['svm'] = np.sqrt(resampled['x']**2 + resampled['y']**2 + resampled['z']**2)
resampled['vmm'] = resampled[["x", "y", "z"]].max(axis=1)
resampled['sma'] = resampled[['x', 'y', 'z']].abs().sum(axis=1)
resampled['ai'] = resampled['svm'] / resampled['n']

# 输出结果
print(resampled)
resampled.to_csv(f"data/resampled/resampled_{type}_{bin}.csv", index=True)

额外优化建议 #

• 时间范围过滤：如果数据集时间跨度大但有效数据集中在某段，可先过滤时间范围，减少resample处理的时间区间。
• 数据类型优化：加载数据时指定列的dtype，比如x/y/z用float32，减少内存占用，提升计算速度。
• 替代resample：若不需要严格的连续时间槽，可用df.groupby(pd.Grouper(key='date', freq=bin, dropna=True))，聚合逻辑一致且可能更高效。

内容的提问来源于stack exchange，提问作者A.L.