基于时间的滑动窗口和测量（改变）数据到达率

Question

我正在尝试实现基于时间的滑动窗口（在 Python 中），即数据源插入新数据项，并且自动删除早于 1h 的项。最重要的是，我需要测量速率，或者更确切地说是数据源插入项目的速率变化。

我的问题有两个方面。首先，如何实现基于时间的窗口的最佳方式。在我目前可能很幼稚的解决方案中，我只是使用 Python 列表window = []。如果是新数据item，我会在项目后面附加当前时间戳：window.append((current_time, item))。然后，使用计时器，每 1 秒我 pop 所有第一个元素的时间戳早于当前 (timestamp-1h)：

threshold = int(time.time()*1000) - self.WINDOW_SIZE_IN_MS
while True:
    try:
        if window[0][0] < threshold:
            del self.word_lists[0]
        else:
            break            
    except:
        break

虽然这可行，但我想知道是否有更聪明的解决方案。

更重要的是，什么是衡量数据项进入窗口的速率变化的好方法。在这里，我不知道如何解决这个问题，至少没有一个听起来也很有效。我的想法很天真：我将 1h 窗口分成 20 个间隔，每 5 分钟一次，并计算项目数。如果最近的 5 分钟间隔包含的项目明显多于 20 间隔的平均值，我说有爆发。但我必须每隔一分钟做一次。这听起来效率不高，而且有很多参数。

简而言之，我需要测量新项目进入我的窗口的加速度。有这方面的最佳实践方法吗？

Answer 1

对于第一部分，当您收到要添加的新项目时，检查过期项目并删除它们会更有效。也就是说，不要为每秒钟无缘无故唤醒进程的计时器烦恼——只要在实际工作发生时搭载维护工作即可。

对于第二部分，整个 1 小时的长度是已知的。存储一个整数，它是五分钟前列表中的索引。您可以在进行插入时保持这一点，并且您知道您只需将其向前移动。

把它们放在一起，伪代码：

window = []
recent_index = 0
def insert(time, item):
    while window and window[0][0] < time - timedelta(hours=1):
        window.pop()
        recent_index -= 1

    while window[recent_index][0] < time - timedelta(minutes=5):
        recent_index += 1

    window.append((time, item))

    return float(len(window) - recent_index) / len(window)

上面的函数返回过去一小时的项目在过去五分钟内到达的比例。比如说，超过 20% 或 50%，你就会爆发。