如何无循环高效拼接数组？优化大规模Numpy数组加载代码

首先看你原来的嵌套循环代码，在M达到百万级的时候效率肯定拉胯，numpy本身就有很多向量化的批量处理能力，完全可以摆脱循环，把速度提上去一大截。

核心优化方案 #

你的需求是把每个sample{i}.npy里的一维数组（长度41632）堆叠成M×41632的二维数组，最直接的无循环实现方式是批量加载所有数组后一次性合并：

import numpy as np

M = 100  # 替换成你的实际M值，最高支持到1e6
# 用列表推导式批量加载所有npy文件，这比手动写for循环快很多
loaded_arrays = [np.load(f'sample{i}.npy') for i in range(M)]
# 直接转为二维数组，numpy会自动完成堆叠操作
sample = np.array(loaded_arrays)

为什么这方法更快？ #

• 列表推导式是Python底层优化过的语法，比手动写for循环+append的效率高不少
• np.array()是基于C实现的向量化操作，完全避开了Python层面的循环，处理百万级数据的速度会比你原来的代码快几个数量级

百万级M的内存注意事项 #

当M达到一百万时，内存占用会非常可观：

• 如果每个数组是float64类型，单个数组占41632*8=333056字节≈325KB，一百万个数组总内存约为316GB，这远超普通机器的内存容量
• 这种情况下建议分块处理：每次加载一小批文件合并，把分块后的数组保存到磁盘，最后再按需合并分块，避免一次性占满内存

分块处理的示例代码：

import numpy as np

M = 1000000
block_size = 10000  # 每次处理10000个文件，可根据内存调整
num_blocks = M // block_size

for block_idx in range(num_blocks):
    start = block_idx * block_size
    end = start + block_size
    # 加载当前块的所有数组
    block_arrays = [np.load(f'sample{i}.npy') for i in range(start, end)]
    block = np.array(block_arrays)
    # 保存分块到磁盘
    np.save(f'block_{block_idx}.npy', block)
    print(f"完成块 {block_idx+1}/{num_blocks}")

# 后续需要完整数组时，再加载所有分块合并
# full_sample = np.vstack([np.load(f'block_{i}.npy') for i in range(num_blocks)])

额外小技巧 #

如果你的sample{i}.npy文件命名连续，还可以用glob模块批量获取文件名，避免手动生成索引：

import glob
import numpy as np

# 获取所有sample开头的npy文件，排序保证顺序正确
file_paths = sorted(glob.glob('sample*.npy'))
loaded_arrays = [np.load(path) for path in file_paths]
sample = np.array(loaded_arrays)

这样代码更简洁，也不用担心文件名索引出错的问题。

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