每秒推特

Question

这个挑战来自才华横溢：

2013年8月3日，“天空中的日本城堡”每秒钟发布一条推文，打破了Twitter的记录。在经典动画片放映期间，人们一度每秒提交143,199条推文。这个特别的峰值比Twitter的稳定状态高出大约25倍。给定一个表示每秒记录的tweet数量的整数数组和一个整数值K，您的任务是编写一个函数，该函数输出到标准输出(stdout) --该数组中每秒钟记录到的最高tweet数--在它之前的K秒内，请注意您的函数将接收以下参数：tps (表示每秒记录的tweet数量的整数数组)和k，后者是上面提到的整数。数据约束:上面的数组长度将不超过50万个数字效率约束:您的函数将打印请求的结果，并在不到2秒的时间内返回示例输入tps: 6、9、4、4、7、4、1 k: 3输出6 9 9 7 7请注意:上述程序应在2秒以下运行。

我的计划

start=1
for i,j in enumerate(tps):
    if start<=k:
        print max(tps[:start])
        start+=1
    else:
        print max(tps[i-(start)+2:i+1])

此函数用于测试用例的时间超过2秒。

我们怎么能在2秒内完成同样的事情呢？如何对其进行优化以使其更快？

Answer 1

变量名i、j、k都很神秘。此外，命名它们同样意味着它们之间有某种关系，但实际上没有任何关系。i是一个索引，j是一个数据点，k是一个固定的区间大小参数。

在循环中使用初始的start <= k情况会减慢循环的速度，即使它已经达到稳定状态。

抽象一下就好了。将代码打包到函数中是一个很好的习惯，而且计算和输出例程要解耦。

有很多数组切片正在进行。我认为解决方案是使用圆形缓冲器。

def max_over_interval(data, interval_size=1):
    ring_buf = [0] * interval_size
    for i, datum in enumerate(data):
        ring_buf[i % interval_size] = datum
        yield(max(ring_buf))

以上解决方案在数据点至少为0时有效，如果数据代表tweet速率，这应该是一个合理的假设。

下一个优化可能是进行一些缓存，以避免调用max()。

from operator import itemgetter

def max_over_interval(data, interval_size=1):
    ring_buf = [0] * interval_size
    cached_max_pos, cached_max = 0, 0
    for i, datum in enumerate(data):
        j = i % interval_size
        ring_buf[j] = datum

        if datum > cached_max:
            cached_max_pos, cached_max = j, datum
            yield cached_max
        elif j != cached_max_pos:
            yield cached_max
        else:
            cached_max_pos, cached_max = max(enumerate(ring_buf), key=itemgetter(1))
            yield cached_max

def tweets_per_second(tps, k):
    for n in max_over_interval(tps, k):
        print(n)

这位在线法官以某种方式宣称，处理81259元素的样本tps和间隔为31763的样本要花费超过2秒的时间，尽管它在我的机器上大约在1/3秒内完成。

Answer 2

使您的解决方案更好

除非你改变你的算法，否则你不会得到很大的性能提升。无论如何，让我们看看可以做些什么来改进您的代码。

您正在使用变量start。很容易检查以下几个属性是否有效(cf assert语句)：

def solution_0_bis(tps, k):
    start=1
    for i,j in enumerate(tps):
        if start<=k:
            assert i < k
            assert i+1 == start
            print max(tps[:start]),
            start+=1
            assert i+2 == start
        else:
            assert i >= k
            assert start == k+1
            print max(tps[i-(start)+2:i+1]),
    print "done"

因此，因为start可以用其他变量重新表示，所以我们可以去掉它。在替换和移除断言之后，您的代码变成：

def solution_0_bis(tps, k):
    for i,j in enumerate(tps):
        if i<k:
            print max(tps[:i+1]),
        else:
            print max(tps[i-(k+1)+2:i+1]),
    print "done"

在第一个切片表达式中引入0和一些数学运算之后，在条件检查和第二个切片表达式中，代码现在是：

def solution_0_bis(tps, k):
    for i,j in enumerate(tps):
        if i-k+1 <= 0:
            print max(tps[0:i+1]),
        else:
            print max(tps[i-k+1:i+1]),
    print "done"

当你摆脱重复的代码时，魔法就会发生：

def solution_0_bis(tps, k):
    for i,j in enumerate(tps):
        print max(tps[max(0,i-k+1):i+1]),
    print "done"

通过删除对max的调用，您可以很容易地看到我们正在处理的部分列表是什么。(公平地说，我首先编写了这段代码，然后尝试将您的代码简化为这样)。

有了这个之后，您就有了一个简单的解决方案，可以用来比较结果，同时在一个评论中提出的建议中实现不同的算法。