如何在超大np.array中快速查找与另一序列最相似的子序列？

核心原理 #

你要找的是S1中与S2最匹配的子序列，本质是计算每个滑动窗口与S2的距离（比如L1/L2范数），然后取最小距离对应的窗口。直接构建窗口矩阵会爆内存，而卷积运算可以把这个滑动窗口计算转化为并行的GPU操作，既省内存又提速度。

以常用的L2距离为例，对于S1的第i个窗口S1[i:i+L]（L是S2的长度），与S2的L2距离平方可以展开为：

||S1_win - S2||² = sum(S1_win²) + sum(S2²) - 2*sum(S1_win * S2)

这三个项都能通过卷积/滑动窗口高效计算：

• sum(S2²)是固定值，只需计算一次
• sum(S1_win²)可以用全1的卷积核与S1²做1D卷积，得到每个窗口的平方和
• sum(S1_win * S2)等价于S1与翻转后的S2做1D卷积（因为PyTorch/TensorFlow的conv1d默认是互相关，需要翻转核才能得到点积结果）

PyTorch实现代码 #

import torch

def find_best_match(S1, S2, device='cuda'):
    L = len(S2)
    N = len(S1)
    
    # 转成torch张量，移到GPU
    s1 = torch.tensor(S1, dtype=torch.float32, device=device).unsqueeze(0).unsqueeze(0)  # shape (1,1,N)
    s2 = torch.tensor(S2, dtype=torch.float32, device=device).unsqueeze(0).unsqueeze(0)  # shape (1,1,L)
    
    # 计算固定项：sum(S2²)
    sum_s2_sq = torch.sum(s2 ** 2)
    
    # 计算每个S1窗口的平方和：用全1卷积核卷积S1²
    s1_sq = s1 ** 2
    ones_kernel = torch.ones((1, 1, L), dtype=torch.float32, device=device)
    sum_s1_win_sq = torch.nn.functional.conv1d(s1_sq, ones_kernel, padding=0)  # shape (1,1,N-L+1)
    
    # 计算S1窗口与S2的点积：翻转S2作为卷积核
    s2_flipped = torch.flip(s2, dims=[2])
    dot_products = torch.nn.functional.conv1d(s1, s2_flipped, padding=0)  # shape (1,1,N-L+1)
    
    # 计算L2距离平方
    l2_sq = sum_s1_win_sq + sum_s2_sq - 2 * dot_products
    l2_sq = l2_sq.squeeze()  # 转成1D张量
    
    # 找最小距离的索引
    best_idx = torch.argmin(l2_sq).item()
    best_distance = torch.sqrt(l2_sq[best_idx]).item()
    
    return best_idx, best_distance

• 若需要L1距离，可把展开式换成绝对值的和，用torch.nn.functional.conv1d配合绝对值处理
• 若S1实在太大，可分块处理：把S1切成若干段，每段长度略大于L，分别计算后再合并结果，避免一次性占满GPU内存

Keras/TensorFlow实现代码 #

import tensorflow as tf

def find_best_match_tf(S1, S2):
    L = len(S2)
    N = len(S1)
    
    # 转成TensorFlow张量
    s1 = tf.convert_to_tensor(S1, dtype=tf.float32)[tf.newaxis, :, tf.newaxis]  # shape (1,N,1)
    s2 = tf.convert_to_tensor(S2, dtype=tf.float32)[tf.newaxis, :, tf.newaxis]
    
    # 固定项：sum(S2²)
    sum_s2_sq = tf.reduce_sum(tf.square(s2))
    
    # 每个S1窗口的平方和
    s1_sq = tf.square(s1)
    ones_kernel = tf.ones((L, 1, 1), dtype=tf.float32)
    sum_s1_win_sq = tf.nn.conv1d(s1_sq, ones_kernel, stride=1, padding='VALID')  # shape (1,N-L+1,1)
    
    # 点积：翻转S2作为卷积核
    s2_flipped = tf.reverse(s2, axis=[1])
    dot_products = tf.nn.conv1d(s1, s2_flipped, stride=1, padding='VALID')  # shape (1,N-L+1,1)
    
    # 计算L2距离平方
    l2_sq = sum_s1_win_sq + sum_s2_sq - 2 * dot_products
    l2_sq = tf.squeeze(l2_sq)
    
    # 找最优索引
    best_idx = tf.argmin(l2_sq).numpy()
    best_distance = tf.sqrt(l2_sq[best_idx]).numpy()
    
    return best_idx, best_distance

关键注意事项 #

• 距离度量选择：对异常值敏感选L1距离，希望突出大差异选L2距离，两种都能通过卷积改写
• 内存优化：超大规模数组可用float16 dtype减半内存占用（注意精度损失）；分块处理时需在块间保留L-1长度的重叠，避免遗漏跨块窗口
• 卷积核翻转：PyTorch和TensorFlow的1D卷积默认是互相关，必须翻转S2才能得到滑动窗口的点积结果，否则距离计算会出错

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