使用 Accurate Scheduler 模拟时间序列事件

Question

我有一个 API，我需要运行一些测试。我们已经完成了压力和负载测试，但最好的测试方法是运行一些真实的数据。我有一个包含过去几年所有历史数据的事实表。目标是找到该历史记录的繁忙窗口，并针对我们的 API “重放”它。

有没有办法在 Python 中“重放时间序列”数据并模拟 API 请求活动。

输入数据是这样的，一天有几十万行：

TimeStamp                      Input Data
------------------------------------------
2020-01-01 00:00:01:231        ABC 
2020-01-01 00:00:01:456        ABD 
2020-01-01 00:00:01:789        XYZ 
...

我首先想到的是将每一行转换为一个cron-entry，所以当每一行被激活时，它会触发一个对API的请求，并将数据条目作为payload。

但是，这种方法增加了启动 Python 进程的大量开销，并且时间分布被破坏：在一秒钟内，它可能会启动大量进程、加载库等。

有没有办法可以启动一个长时间运行的 Python 进程以根据时间序列数据完美重播？ （最好在几毫秒内准确）

差不多：

while True: 
    currenttime = datetime.now() 
    # find from table rows with currentime 
    # make web requests with those rows 

然后这变得同步，每个循环都需要数据库查找..

Answer 1

也许您希望将您的实时播放例程写成这样（伪代码）：

def playbackEventsInWindow(startTime, endTime):
   timeDiff = datetime.timedelta(startTime, datetime.now()).total_seconds()
   prevTime = startTime
   while True: 
      nextEvent = GetFirstEventInListAfterSpecifiedTime(prevTime)
      if nextEvent:
         nextTime = nextEvent.getEventTimeStamp()
         if (nextTime >= endTime):
            return  # we've reached the end of our window
         sleepTimeSeconds = datetime.timedelta(datetime.now(), nextTime).total_seconds()+timeDiff
         if (sleepTimeSeconds > 0.0):
            time.sleep(sleepTimeSeconds)
         executeWebRequestsForEvent(nextEvent)
         prevTime = nextTime
      else:
         return  # we've reached the end of the list

请注意，GetFirstEventInListAfterSpecifiedTime(timeStamp) 的简单实现将简单地从事件列表的开头开始，然后线性扫描列表，直到找到时间戳大于指定参数的事件，然后返回该事件。 . 但是如果事件列表很长，那么这种实现很快就会变得非常低效。但是，您可以通过让它存储它在前一次调用中返回的值的索引来调整它，并在该位置而不是从列表顶部开始线性扫描。这将允许它在常见情况下（即请求的时间戳稳步增加）快速返回（即通常只需一步）。