在数据流中找到迄今为止的前 K 个最小值答案

【问题标题】：Find Top-K Smallest Values So Far in Data Stream在数据流中找到迄今为止的前 K 个最小值
【发布时间】：2021-09-03 08:19:22
【问题描述】：

假设我有一个数据流，其中一次检索单个数据点：

import numpy as np
def next_data_point():
    """
    Mock a data stream. Data points will always be a positive float
    """
    return np.random.uniform(0, 1_000_000, dtype='float')

我需要能够更新 NumPy 数组并跟踪该流中迄今为止的前 K 个最小值（或者直到用户决定何时可以通过某些 check_stop_condition() 函数停止分析） .假设我们想从流中捕获前 1,000 个最小值，那么实现此目的的简单方法可能是：

k = 1000
topk = np.full(k, fille_value=np.inf, dtype='float')
while check_stop_condition():
    topk[:] = np.sort(np.append(topk, next_data_point()))[:k]

这很好用，但效率很低，如果重复数百万次可能会很慢，因为我们是：

每次都创建一个新数组
每次都对串联的数组进行排序

所以，我想出了一种不同的方法来解决这两个低效率问题：

k = 1000
topk = np.full(k, fille_value=np.inf)
while check_stop_condition():
    data_point = next_data_point()
    idx = np.searchsorted(topk, data_point)
    if idx < k:
        topk[idx : -1] = topk[idx + 1 :] 
        top[idx] = data_point

在这里，我利用np.searchsorted() 替换np.sort 并快速找到下一个数据点的插入点idx。我相信np.searchsorted 使用某种二进制搜索并假设初始数组首先被预先排序。然后，我们在topk 中移动数据以适应并插入新数据点当且仅当idx < k。

我还没有看到任何地方都这样做过，所以我的问题是，是否可以做些什么来提高效率？尤其是在我在 if 语句中进行转换的方式。

【问题讨论】：

标签： python arrays performance numpy

【解决方案1】：

对一个巨大的数组进行排序非常昂贵，因此第二种方法更快也就不足为奇了。但是，第二种方法的速度可能受到慢速数组复制的限制。第一种方法的复杂度为O(k log(k) n)，第二种方法的复杂度为O(n (log(k) + k * p))，其中n是点数，p是要采取分支的概率。

要构建更快的实现，您可以使用树。更具体地说，例如 self-balancing binary search tree。这是算法：

topk = Tree()
maxi = np.inf
while check_stop_condition():             # O(n)
    data_point = next_data_point()
    if len(topk) <= 1000:                 # O(1)
        topk.insert(data_point)           # O(log k)
    elif data_point < maxi:               # Discard the value in O(1)
        topk.insert(data_point)           # O(log k)
        topk.deleteMaxNode()              # O(log k)
        maxi = topk.findMaxValue()        # O(log k)

上述算法在O(n log k) 中运行。可以证明这种复杂性是最优的（仅使用 data_point 比较）。

在实践中，二进制堆可能会更快一些（具有相同的复杂性）。实际上，在这种情况下，它们比自平衡二叉搜索树有几个优势：

它们可以在内存中以非常紧凑的方式实现（减少缓存未命中和内存消耗）
插入n=1000 第一项可以在O(n) 时间内完成，而且速度非常快

请注意，丢弃的值是在恒定时间内计算的。这会在巨大的随机数据集上附加很多内容，因为大多数值会迅速大于maxi。 On 甚至可以证明随机数据集可以在O(n)时间计算（最优）。

请注意，Python 3 提供了一个名为 heapq 的标准堆实现，这可能是一个很好的起点。

【讨论】：