寻求Python实现长表转宽表并按1分钟间隔聚合的优化方案

需求背景 #

需要将时间戳不一致的长格式传感器数据，转换为以1分钟为固定间隔的宽格式表，每个通道的val1、val2单独列为一列。

原始输入数据 #

channelId,utc,scet,val1,val2
A-0001,2024-061T22:00:05.02064,0.03,3,
A-0002,2024-061T22:00:06.02064,0.07,2,
A-0001,2024-061T22:00:11.02064,0.02,2,
A-0002,2024-061T22:00:12.02064,0.05,7,
A-0001,2024-061T22:01:12.365611,0.01,1.5,
A-0002,2024-061T22:01:14.365611,.07,16

现有实现代码 #

import pandas as pd

# Read the input table into a DataFrame
df = pd.read_csv('~/Desktop/test_file_1.csv')

# Convert timestamp columns to datetime format with explicit format specification
df['utc'] = pd.to_datetime(df['utc'], format='%Y-%jT%H:%M:%S.%f')

# Round timestamps to the nearest minute
df['utc'] = df['utc'].dt.round('min')

# Pivot the DataFrame
df_pivot = df.pivot_table(index=['utc'], columns='channelId', values=['val1', 'val2'])

df_reset = df_pivot.reset_index()
df_reset['utc'] = pd.to_datetime(df_reset['utc'])
df_reset.set_index('utc', inplace=True)

# Resample the DataFrame to get values for every minute
df_resampled = df_reset.resample('T').last().ffill()

# Flatten multi-level column index
df_resampled.columns = [f'{col[1]}_{col[0]}' for col in df_resampled.columns.values]

# Reset index
df_resampled.reset_index(inplace=True)

# Rename columns
df_resampled.rename(columns={'ert': 'Time'}, inplace=True)

df_final = df_resampled[['Time', *sorted(df_resampled.columns[1:])]]

# Write the output table to a CSV file
df_final.to_csv('output_table_3.csv', index=False)

现有输出结果（注：原输出中22:00:01为笔误，实际应为22:01:00） #

Time, A-0001_val1, A-0001_val2, A-0002_val1, A-0002_val2
2024-03-01 22:00:00,0.02,2,0.05,7
2024-03-01 22:01:00,0.01,1.5,0.07,16

优化后的实现代码 #

import pandas as pd

# 读取数据时直接解析UTC时间，避免后续重复转换
df = pd.read_csv(
    '~/Desktop/test_file_1.csv',
    parse_dates=['utc'],
    date_parser=lambda x: pd.to_datetime(x, format='%Y-%jT%H:%M:%S.%f')
)

# 将时间戳取整到分钟级别，并设置为索引
df['utc'] = df['utc'].dt.round('min')
df.set_index('utc', inplace=True)

# 合并透视、重采样、填充步骤，减少中间变量
df_wide = df.pivot(columns='channelId', values=['val1', 'val2']) \
            .resample('T').last() \
            .ffill()

# 扁平化多级列名，格式为"通道_字段"
df_wide.columns = df_wide.columns.map(lambda x: f"{x[1]}_{x[0]}")

# 重置索引、修改列名并排序列
df_final = df_wide.reset_index() \
                  .rename(columns={'utc': 'Time'}) \
                  .reindex(columns=['Time'] + sorted(df_wide.columns))

# 输出到CSV
df_final.to_csv('output_table_3.csv', index=False)

优化点说明 #

• 减少重复操作：读取数据时直接解析UTC时间，避免后续多次调用pd.to_datetime
• 简化流程：合并透视、重采样、填充步骤，减少中间DataFrame变量，提升代码可读性和运行效率
• 修正原代码错误：原代码中rename(columns={'ert': 'Time'})为笔误，优化后直接修改正确的utc列名为Time
• 更简洁的列名处理：用map替代列表推导式处理多级列名，代码更简洁
• 避免冗余操作：去掉不必要的reset_index和set_index来回切换，直接一步到位

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