如何mmap行式C结构体二进制大文件为Pandas/Dask DataFrame

问题1：numpy memmap 映射多列实现 #

numpy 原生支持结构化类型的内存映射，无需手动指定单字段的步长、偏移，只要定义和C结构体内存布局完全一致的结构化dtype，映射后取字段即可得到零拷贝的列视图，完全满足步长、偏移的要求。

实现步骤 #

• 首先定义匹配C结构体的numpy结构化dtype，注意对齐规则和字节序要和文件写入时完全一致：

import numpy as np
import pandas as pd
import os

# 对应示例结构体，小端序、无内存填充，和struct.pack('<idh')的布局完全匹配
struct_dtype = np.dtype([
    ('a', '<i4'),  # int32_t a，偏移0，长度4字节
    ('b', '<f8'),  # double b，偏移4，长度8字节
    ('c', '<i2')   # int16_t c，偏移12，长度2字节
], align=False)

print(struct_dtype.itemsize)  # 输出14，和预期结构体步长一致

注意：如果你的C结构体是编译器默认对齐的（没有用packed属性），把align参数设为True，numpy会自动计算填充字节，保证布局匹配。

• 直接对整个文件做内存映射：

total_rows = os.path.getsize("binarydb") // struct_dtype.itemsize
# 内存映射文件，不会把全量数据加载到物理内存
mmapped_struct_arr = np.memmap(
    filename="binarydb",
    dtype=struct_dtype,
    mode="r",  # 按需修改为r+、w+等读写模式
    offset=0,  # 如果文件有头部信息，修改为头部字节长度
    shape=(total_rows,)
)

• 提取各列为numpy数组，这些数组都是零拷贝的内存映射视图，步长自动为结构体大小，偏移自动匹配字段位置：

col_a = mmapped_struct_arr['a']  # int32数组，偏移0，步长14
col_b = mmapped_struct_arr['b']  # float64数组，偏移4，步长14
col_c = mmapped_struct_arr['c']  # int16数组，偏移12，步长14

• 对接pandas DataFrame，直接传入列视图即可，不会触发全量数据拷贝：

df = pd.DataFrame({
    'a': col_a,
    'b': col_b,
    'c': col_c
})

问题2：集成Dask实现惰性分页加载 #

Dask支持通过自定义延迟读取函数对接这类结构化二进制文件，全程惰性加载，自动分块调度，不会把全量文件读入内存，适配数亿行级别的大文件场景。

实现步骤 #

• 配置分块参数，定义块读取逻辑：

import dask.array as da
import dask.dataframe as dd

# 配置单块行数，根据可用内存调整，示例为单块100万行
block_row_count = 1_000_000

# 定义单块读取函数，仅在计算任务触发时实际执行
def load_block(start_row, end_row):
    byte_offset = start_row * struct_dtype.itemsize
    block_len = end_row - start_row
    return np.memmap(
        "binarydb",
        dtype=struct_dtype,
        mode="r",
        offset=byte_offset,
        shape=(block_len,)
    )

• 构建延迟分块任务，生成Dask DataFrame：

# 生成分块的延迟读取任务列表
delayed_blocks = [
    da.delayed(load_block)(s, min(s+block_row_count, total_rows))
    for s in range(0, total_rows, block_row_count)
]

# 构建Dask结构化数组，再转换为Dask DataFrame
dask_arr = da.from_delayed(
    delayed_blocks,
    shape=(total_rows,),
    dtype=struct_dtype
)
dask_df = dd.from_dask_array(dask_arr, columns=['a', 'b', 'c'])

生成的dask_df支持和pandas一致的API，做筛选、聚合、分组等操作时，Dask只会读取需要的数据块。如果需要更高的并行性能，可以把大文件拆分为多个等大小的分片，Dask可以并行读取多个分片提升吞吐量。

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