如何实现内存高效的滑动窗口计算？含NumPy与CuPy优化问询

NumPy 实现（CPU平台） #

针对GB级大数组，核心是避免生成完整的多通道邻域数组，通过逐个方向比较并跟踪最大值及对应方向，将内存占用控制在最低水平：

思路 #

1. 定义8个相邻方向的偏移量（上下左右、四个对角线）。
2. 对每个方向，提取与原数组中心区域对齐的子数组（均为原数组的视图，无内存复制）。
3. 初始化最大值数组和方向索引数组，遍历所有方向，逐元素比较并更新最大值与对应方向。

代码示例 #

import numpy as np

# 假设输入为L×L的大数组
L = 10000
arr = np.random.rand(L, L).astype(np.float64)  # 示例数组，实际为你的GB级数据

# 定义8个相邻方向的偏移（行偏移，列偏移），顺序对应方向索引0-7
dirs = [(-1, -1), (-1, 0), (-1, 1),
        (0, -1),          (0, 1),
        (1, -1),  (1, 0), (1, 1)]

# 中心区域切片（排除边界元素，因为边界无完整邻域）
y_center = slice(1, -1)
x_center = slice(1, -1)

# 初始化最大值和方向索引：取第一个方向的子数组作为初始最大值
dy0, dx0 = dirs[0]
max_vals = arr[y_center.start + dy0 : y_center.stop + dy0,
               x_center.start + dx0 : x_center.stop + dx0]
# 用int8存储方向索引，进一步节省内存
dir_indices = np.zeros_like(max_vals, dtype=np.int8)

# 遍历剩余方向，逐元素更新最大值与方向
for idx in range(1, len(dirs)):
    dy, dx = dirs[idx]
    current_neighbors = arr[y_center.start + dy : y_center.stop + dy,
                            x_center.start + dx : x_center.stop + dx]
    # 找到当前方向值大于现有最大值的位置
    update_mask = current_neighbors > max_vals
    # 更新最大值和方向索引
    max_vals[update_mask] = current_neighbors[update_mask]
    dir_indices[update_mask] = idx

# dir_indices即为每个中心元素的最大邻居方向索引，对应dirs中的顺序

内存优势 #

此方法仅需存储两个与中心区域（(L-2)×(L-2)）大小相同的数组，相比生成8/9通道的邻域数组，内存占用降低75%以上，完全适配GB级数组的处理需求。

CuPy 实现（GPU平台） #

CuPy的最优策略需结合GPU内存容量和并行特性调整，核心同样是避免不必要的内存复制：

场景1：GPU内存充足 #

利用GPU并行处理优势，直接堆叠所有方向的子数组后调用argmax，代码简洁且效率极高：

import cupy as cp

L = 10000
arr = cp.random.rand(L, L).astype(cp.float64)  # GPU上的数组

dirs = [(-1, -1), (-1, 0), (-1, 1),
        (0, -1),          (0, 1),
        (1, -1),  (1, 0), (1, 1)]

y_center = slice(1, -1)
x_center = slice(1, -1)

# 堆叠所有方向的子数组为最后一维（GPU上的视图堆叠，内存开销可控）
neighbors_stack = cp.stack([
    arr[y_center.start + dy : y_center.stop + dy,
        x_center.start + dx : x_center.stop + dx]
    for dy, dx in dirs
], axis=-1)

# 一次性计算每个位置的最大方向索引
dir_indices = cp.argmax(neighbors_stack, axis=-1)

场景2：GPU内存紧张 #

采用与NumPy一致的逐个方向比较更新策略，仅占用少量GPU内存：

import cupy as cp

L = 10000
arr = cp.random.rand(L, L).astype(cp.float64)

dirs = [(-1, -1), (-1, 0), (-1, 1),
        (0, -1),          (0, 1),
        (1, -1),  (1, 0), (1, 1)]

y_center = slice(1, -1)
x_center = slice(1, -1)

dy0, dx0 = dirs[0]
max_vals = arr[y_center.start + dy0 : y_center.stop + dy0,
               x_center.start + dx0 : x_center.stop + dx0]
dir_indices = cp.zeros_like(max_vals, dtype=cp.int8)

for idx in range(1, len(dirs)):
    dy, dx = dirs[idx]
    current_neighbors = arr[y_center.start + dy : y_center.stop + dy,
                            x_center.start + dx : x_center.stop + dx]
    update_mask = current_neighbors > max_vals
    # 用CuPy的where实现向量化更新
    max_vals = cp.where(update_mask, current_neighbors, max_vals)
    dir_indices = cp.where(update_mask, idx, dir_indices)

关键注意点 #

• CuPy原生支持向量化操作，无需依赖Numba（Numba对CuPy的支持有限且复杂度高）。
• 所有操作均在GPU上完成，避免主机与设备间的数据传输开销。

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