Polars中二进制数据高效解码方案问询：替代map_elements的优化方法

我在处理二进制数据时，用自定义函数结合Polars的map_elements方法解码，但这种逐元素处理的方式在大数据集下耗时极高。尝试把二进制数据转成List(UInt8)类型时失败，报错如下：

exceptions.ArrowErrorException: NotYetImplemented("Casting from LargeBinary to LargeList(Field { name: \"item\", data_type: UInt8, is_nullable: true, metadata: {} }) not supported")

有没有更高效的解码方式？

相关代码示例及执行结果 #

创建DataFrame

import polars as pl
import struct
import io

data = {"binary": [b'\xFD\x00\xFE\x00\xFF\x00',b'\x10\x00\x20\x00\x30\x00'], "id": [1,2]}
schema = {"binary": pl.Binary, "id":pl.Int16}

df = pl.DataFrame(data, schema)

执行结果：

shape: (2, 2)
┌───────────────┬─────┐
│ binary        ┆ id  │
│ ---           ┆ --- │
│ binary        ┆ i16 │
╞═══════════════╪═════╡
│ [binary data] ┆ 1   │
│ [binary data] ┆ 2   │
└───────────────┴─────┘

应用自定义解码函数

def custom_decode(data):
   bytestream = io.BytesIO(data)
   lst = []

   while bytestream.tell() < 6:
      lst.append(struct.unpack('<H', bytestream.read(2))[0])

   return lst

df = df.with_columns(
      pl.col('binary').map_elements(lambda x: custom_decode(x))
   )

执行结果：

shape: (2, 2)
┌─────────────────┬─────┐
│ binary          ┆ id  │
│ ---             ┆ --- │
│ list[i64]       ┆ i16 │
╞═════════════════╪═════╡
│ [253, 254, 255] ┆ 1   │
│ [16, 32, 48]    ┆ 2   │
└─────────────────┴─────┘

高效解决方案 #

方案1：Polars原生向量化解析（推荐）

直接用Polars的二进制结构解析能力，全程在Arrow层面批量处理，无需Python循环：

df = df.with_columns(
    pl.col("binary")
    .binary.to_struct(
        [("v1", pl.UInt16), ("v2", pl.UInt16), ("v3", pl.UInt16)],
        byte_order="little"
    )
    .struct.to_list()
    .alias("decoded")
)

方案2：批量转换为字节列表后处理

如果二进制长度不固定，先批量转成字节列表，再用窗口操作解析：

# 批量转成UInt8列表
df = df.with_columns(
    pl.col("binary")
    .map_batches(
        lambda s: pl.Series([list(x) for x in s.to_numpy()]),
        return_dtype=pl.List(pl.UInt8)
    )
    .alias("bytes_list")
)

# 按2字节窗口解析小端序UInt16
df = df.with_columns(
    pl.col("bytes_list")
    .list.eval(
        (pl.element().list.get(1) << 8) + pl.element().list.get(0),
        strategy="window",
        window_size=2
    )
    .alias("decoded")
)

方案3：结合numpy批量解码

利用numpy的C级解析能力，一次性处理所有二进制数据：

import numpy as np

def batch_decode(binary_series):
    # 拼接所有二进制数据，批量解析为小端序UInt16
    arr = np.frombuffer(b''.join(binary_series.to_list()), dtype='<u2')
    # 按原字段长度拆分（6字节对应3个数值）
    return arr.reshape(-1, 3).tolist()

df = df.with_columns(
    pl.Series(batch_decode(df["binary"])).alias("decoded")
)

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