如何提升Pandas-NumPy解码大字节序列的性能？

问题背景 #

需处理数十个二进制文件，每个文件包含数百个信号，每个信号约10^5个样本，每个样本为17字节的固定结构：

• time: uint64类型（以0001/01/01 00:00:00为起点的100ns单位计数）
• value: float64类型
• flags: uint8类型

目标是将时间字段转换为datetime64[ns]格式，最终存入Pandas DataFrame或NumPy数组。现有实现处理100个信号耗时5.5-6.7秒，性能未达预期。

核心优化思路 #

1. 用NumPy直接解析二进制数据，替代struct模块 #

struct.unpack/iter_unpack会先将二进制数据转换为Python原生类型的元组/列表，再转NumPy数组，存在额外的类型转换开销。使用numpy.frombuffer可直接在内存上构建NumPy数组，跳过中间环节，大幅提升效率。

2. 避免不必要的浮点运算，直接用整数运算转换时间 #

原实现中时间转换涉及浮点减法和乘法，可直接通过整数运算将100ns单位转换为ns单位，再直接转datetime64[ns]，既减少运算耗时又避免浮点精度损失。

3. 结构化数组一次性存储所有字段 #

用NumPy结构化数组直接存储三个字段，减少多次数组拆分的开销，后续转Pandas DataFrame也更便捷。

优化后的代码实现 #

import pandas as pd
import numpy as np
import timeit
from struct import unpack, iter_unpack

# DELTA_EPOCH is the delay in 100 ns units from 0001/01/01 to 1970/01/01.
DELTA_EPOCH = 621355968000000000

# MAGIC_NUMBER is the factor to convert from 100 ns units to seconds.
MAGIC_NUMBER = 1/10000000
NS_PER_100NS = 100

def current_implementation():
    fmt = "<" + "QdB" * nsamples
    nums = unpack(fmt, signal)
    ts = np.array(nums[::3], dtype=np.float64)
    vl = np.array(nums[1::3], dtype=np.float64)
    fl = np.array(nums[2::3], dtype=np.uint8)
    ts = (ts - DELTA_EPOCH) * MAGIC_NUMBER
    idx = pd.to_datetime(ts, unit='s')
    return idx, vl, fl

def alternative_implementation1():
    fmt = "<QdB"
    tmp = list(iter_unpack(fmt, signal))
    tmp = np.array(tmp, dtype=[('idx', np.uint64), ('vl', np.float64), ('fl', np.uint8)])
    tmp['idx'] = np.array((tmp['idx'] - DELTA_EPOCH) * 100, dtype = 'datetime64[ns]')
    return tmp

def alternative_implementation2():
    fmt = "<" + "QdB" * nsamples
    nums = unpack(fmt, signal)
    tmp = np.array(nums[::3], dtype=np.uint64) - DELTA_EPOCH
    ts = np.array(tmp * 100, dtype='datetime64[ns]')
    vl = np.array(nums[1::3], dtype=np.float64)
    fl = np.array(nums[2::3], dtype=np.uint8)
    idx = pd.DatetimeIndex(ts)
    return idx, vl, fl

def optimized_implementation():
    # 定义小端字节序的结构化数据类型
    dtype = np.dtype([('time', '<u8'), ('value', '<f8'), ('flags', 'u1')])
    # 直接从二进制缓冲区加载为结构化数组，无额外内存复制
    arr = np.frombuffer(signal, dtype=dtype)
    
    # 整数运算转换时间：转成ns单位后直接转为datetime64[ns]
    arr['time'] = (arr['time'] - DELTA_EPOCH) * NS_PER_100NS
    arr['time'] = arr['time'].astype('datetime64[ns]')
    
    # 转成Pandas对象返回
    df = pd.DataFrame(arr)
    return df['time'], df['value'], df['flags']

def optimized_implementation_no_pandas():
    # 纯NumPy处理，无需Pandas转换
    dtype = np.dtype([('time', '<u8'), ('value', '<f8'), ('flags', 'u1')])
    arr = np.frombuffer(signal, dtype=dtype)
    
    arr['time'] = (arr['time'] - DELTA_EPOCH) * NS_PER_100NS
    arr['time'] = arr['time'].astype('datetime64[ns]')
    
    return arr['time'], arr['value'], arr['flags']

if __name__ == "__main__":
    nsamples = 300000
    signal = b'\xcd\xf1\xb9!\x18\xbb\xda\x08\x00\x00\x00\x80\x01\xc84@\x03' * nsamples

    print("Current implementation ", end="")
    print(timeit.timeit('current_implementation()', number=100, globals=globals()))
    
    print("\nalternative_implementation1 ", end="")
    print(timeit.timeit('alternative_implementation1()', number=100, globals=globals()))
    
    print("\nalternative_implementation2 ", end="")
    print(timeit.timeit('alternative_implementation2()', number=100, globals=globals()))
    
    print("\noptimized_implementation ", end="")
    print(timeit.timeit('optimized_implementation()', number=100, globals=globals()))
    
    print("\noptimized_implementation_no_pandas ", end="")
    print(timeit.timeit('optimized_implementation_no_pandas()', number=100, globals=globals()))

额外性能提升建议 #

• 批量读取文件：如果文件是连续的二进制块，可一次性读取多个信号甚至整个文件，再按信号长度拆分，减少IO次数。
• 多进程并行处理：利用multiprocessing模块并行处理多个文件或信号，充分利用CPU多核资源（CPU密集型任务优先用多进程，规避GIL限制）。
• 避免不必要的Pandas转换：若最终仅需NumPy数组，跳过Pandas的DatetimeIndex转换，直接使用NumPy原生datetime64[ns]类型即可。

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