大Pandas DataFrame重采样遇ArrayMemoryError问题求解

问题核心 #

600-700万行13列的传感器CSV日志，仅首尾行含全量传感器值，中间大量NaN，时间步长不固定；需转换为1s固定步长、全值的数据集供仿真工具使用。此前尝试Pandas分块处理耗时1.5小时，Dask在resample("1ms").asfreq()步骤因生成巨量1ms时间序列爆内存（需58.6GiB），且Dask无直接替代asfreq()的函数完成后续插值。

优化方案 #

方案1：跳过1ms中间步长，直接插值到目标1s步长 #

核心思路：无需生成1ms级中间数据集，直接基于原始时间戳的有效数据，插值到1s目标时间点，从根源避免内存爆炸。

import pandas as pd
from scipy.interpolate import interp1d

# 1. 高效加载数据，指定数据类型压缩内存
sensor_cols = ['温度', '压力', '速度']  # 替换为你的13个传感器列
dtype_spec = {col: 'float32' for col in sensor_cols}
df = pd.read_csv(
    'sensor_log.csv',
    parse_dates=['时间戳'],
    dtype=dtype_spec,
    index_col='时间戳'
)

# 2. 确保时间序列严格有序
df = df.sort_index()

# 3. 生成目标1s时间序列
start_ts = df.index.min().floor('1s')
end_ts = df.index.max().ceil('1s')
target_ts = pd.date_range(start=start_ts, end=end_ts, freq='1s')

# 4. 逐列插值补全
interpolated_data = {}
for col in sensor_cols:
    # 提取当前列的有效非NaN数据
    valid_series = df[col].dropna()
    # 转换时间戳为int64数值（scipy插值需要数值型输入）
    valid_times = valid_series.index.astype('int64')
    valid_values = valid_series.values
    
    # 线性插值（支持首尾值外推）
    f = interp1d(
        valid_times,
        valid_values,
        kind='linear',
        fill_value="extrapolate",
        assume_sorted=True
    )
    interpolated_data[col] = f(target_ts.astype('int64'))

# 生成最终数据集
result_df = pd.DataFrame(interpolated_data, index=target_ts)

方案2：Dask分块处理，手动生成目标时间序列 #

针对超大数据集，用Dask分块处理原始数据，每个块独立生成对应时间范围的1s目标点并插值，最后合并去重。

import dask.dataframe as dd
import pandas as pd
from scipy.interpolate import interp1d

# 1. 分块加载数据
sensor_cols = ['温度', '压力', '速度']
dtype_spec = {col: 'float32' for col in sensor_cols}
ddf = dd.read_csv(
    'sensor_log.csv',
    parse_dates=['时间戳'],
    dtype=dtype_spec,
    blocksize='100MB'  # 按需调整分块大小
)

# 2. 定义分块处理函数
def process_chunk(chunk):
    chunk = chunk.set_index('时间戳').sort_index()
    if chunk.empty:
        return pd.DataFrame()
    
    # 生成当前块对应的1s目标时间序列
    start_ts = chunk.index.min().floor('1s')
    end_ts = chunk.index.max().ceil('1s')
    target_ts = pd.date_range(start=start_ts, end=end_ts, freq='1s')
    
    interpolated = {}
    for col in sensor_cols:
        valid_series = chunk[col].dropna()
        if len(valid_series) < 2:
            continue  # 无足够有效数据时跳过，后续可统一补全首尾值
        
        valid_times = valid_series.index.astype('int64')
        valid_values = valid_series.values
        f = interp1d(valid_times, valid_values, kind='linear', fill_value="extrapolate")
        interpolated[col] = f(target_ts.astype('int64'))
    
    return pd.DataFrame(interpolated, index=target_ts)

# 3. 分块执行并合并结果
result_ddf = ddf.map_partitions(
    process_chunk,
    meta=pd.DataFrame(columns=sensor_cols, index=pd.DatetimeIndex([]), dtype='float32')
)
result_df = result_ddf.compute().sort_index()

# 去重重叠时间点（分块可能产生重复）
result_df = result_df[~result_df.index.duplicated(keep='first')]

方案3：Vaex内存映射处理（适合超大规模文件） #

Vaex采用内存映射技术，无需加载全量数据到内存，直接处理磁盘上的CSV文件，内存占用极低。

import vaex
import pandas as pd
from scipy.interpolate import interp1d

# 1. 内存映射加载CSV（不占用内存）
vx_df = vaex.read_csv('sensor_log.csv', parse_dates=['时间戳'])

# 2. 转换时间戳为数值型用于插值
vx_df['time_int'] = vx_df['时间戳'].astype('datetime64[ns]').astype('int64')

# 3. 生成目标1s时间序列
start_ts = vx_df['时间戳'].min().floor('1s')
end_ts = vx_df['时间戳'].max().ceil('1s')
target_ts = pd.date_range(start=start_ts, end=end_ts, freq='1s')
target_time_int = target_ts.astype('int64')

# 4. 逐列插值
sensor_cols = ['温度', '压力', '速度']
result_data = {}
for col in sensor_cols:
    # 提取有效数据
    valid_mask = ~vx_df[col].isna()
    valid_times = vx_df[valid_mask]['time_int'].values
    valid_values = vx_df[valid_mask][col].values
    
    # 线性插值
    f = interp1d(valid_times, valid_values, kind='linear', fill_value="extrapolate")
    result_data[col] = f(target_time_int)

# 生成最终数据集
result_df = pd.DataFrame(result_data, index=target_ts)

关键优化总结 #

1. 避免中间巨量数据集：跳过1ms步长的生成，直接插值到目标1s步长，这是解决内存问题的核心。
2. 数据类型压缩：用float32替代float64存储传感器数值，可减少50%的内存占用。
3. 分块/内存映射：超大规模数据优先用Dask分块或Vaex内存映射，避免全量加载。

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