单遍算法寻找前X百分比项目

Question

我在寻找一种算法，可以在不知道总共有多少个浮点数的情况下，从一个数据流中找到前X%的浮点数。这个数据流的浮点数大概在500万到3000万之间。因为数据是实时生成的，所以我需要一种单次遍历的算法，不能再重新生成一次数据流。

我现在想到的算法是保持一个已经排序好的列表，里面存放我目前看到的前X个浮点数。随着数据流的继续，我会根据需要扩大这个列表。然后，我会用bisect_left来找到插入点，如果需要的话。

下面是我目前的算法：

from bisect import bisect_left
from random import uniform
from itertools import islice


def data_gen(num):
    for _ in xrange(num):
        yield uniform(0,1)

def get_top_X_percent(iterable, percent = 0.01, min_guess = 1000):

    top_nums = sorted(list(islice(iterable, int(percent*min_guess)))) #get an initial guess

    for ind, val in enumerate(iterable, len(top_nums)):
        if int(percent*ind) > len(top_nums):
            top_nums.insert(0,None)
        newind = bisect_left(top_nums, val)
        if newind > 0:
            top_nums.insert(newind, val)
            top_nums.pop(0)

    return top_nums

if __name__ == '__main__':

    num = 1000000
    all_data = sorted(data_gen(num))
    result = get_top_X_percent(all_data)
    assert result[0] == all_data[-int(num*0.01)], 'Too far off, lowest num:%f' % result[0] 
    print result[0]

在实际情况下，这些数据并不是来自任何标准的分布（要是能的话，我就可以用一些统计知识了）。

任何建议都非常欢迎。

Answer 1

正如其他回答所提到的，存储整个数据流在内存中是比较有效的做法。考虑一下这种方式，毕竟5到3000万的浮点数大约只占用40到240MB的内存，这个大小是可以接受的。

既然你要存储整个数据流，那么最快的方法来找出前X%的数据，就是先找到一个分界值（也就是在前X%中最小的那个数），可以用一种线性时间的选择算法来实现：

http://en.wikipedia.org/wiki/Selection_algorithm

然后，再遍历一遍数据流，把所有比这个分界值小的数过滤掉。

这种方法的时间和空间复杂度都是线性的，这是你能得到的最好的效果。

Answer 2

你必须把整个数据流都存到内存里。

证明：假设你有一串数字，n₁,…,n_k。这里的k是个未知数。你怎么知道什么时候可以忘掉 n_i 这个数字呢？当你看到超过 n_i 的数字有 x*k/100 个的时候。但是，因为k是未知的，所以你永远无法做到这一点。

所以，唯一的“单次遍历”算法必须把整个序列都存到内存里。

Answer 3

我不太确定有没有可靠的方法来做到这一点，因为“前X百分比”所代表的范围会随着你看到的元素数量而不可预测地变化。想象一下下面这个输入：

 101 102 103 104 105 106 107 108 109 110 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0...

如果你想要前25%的元素，你可能会从前十个元素中选出101和102，但当你看到足够多的零之后，最终你可能不得不选出前十个元素中的所有。这种情况可以扩展到任何足够大的数据流中——总是有可能因为表面现象而被误导，丢弃那些其实应该保留的元素。因此，除非你事先知道数据流的确切长度，否则我认为这是不可能的（除非你在数据流结束之前把每个元素都保存在内存中）。