来自非常长的可迭代的随机样本，在 python 中

Question

我有一个很长的 python 生成器，我想通过随机选择值的子集来“精简”它。不幸的是，random.sample() 不适用于任意迭代。显然，它需要支持len() 操作的东西（可能是对序列的非顺序访问，但这还不清楚）。而且我不想建立一个庞大的列表，以便我可以精简它。

事实上，在不知道其长度的情况下，可以一次均匀地从序列中采样——Programming perl 中有一个很好的算法可以做到这一点（编辑： “水库采样”，感谢@user2357112！）。但是有人知道提供此功能的标准 python 模块吗？

问题演示（Python 3）

>>> import itertools, random
>>> random.sample(iter("abcd"), 2)
...
TypeError: Population must be a sequence or set.  For dicts, use list(d).

在 Python 2 上，错误更加透明：

Traceback (most recent call last):
  File "<pyshell#12>", line 1, in <module>
    random.sample(iter("abcd"), 2)
  File "/usr/local/Cellar/python/2.7.9/Frameworks/Python.framework/Versions/2.7/lib/python2.7/random.py", line 321, in sample
    n = len(population)
TypeError: object of type 'iterator' has no len()

如果没有random.sample() 的替代品，我会尝试将生成器包装到提供__len__ 方法的对象中（我可以提前找出长度）。所以我会接受一个显示如何干净地做到这一点的答案。

Answer 1

由于您知道可迭代对象返回的数据长度，您可以使用xrange() 快速生成可迭代对象的索引。然后你可以运行迭代器，直到你获取所有数据：

import random

def sample(it, length, k):
    indices = random.sample(xrange(length), k)
    result = [None]*k
    for index, datum in enumerate(it):
        if index in indices:
            result[indices.index(index)] = datum
    return result

print sample(iter("abcd"), 4, 2)

或者，这里是使用“算法 R”的水库采样实现：

import random

def R(it, k):
    '''https://en.wikipedia.org/wiki/Reservoir_sampling#Algorithm_R'''
    it = iter(it)
    result = []
    for i, datum in enumerate(it):
        if i < k:
            result.append(datum)
        else:
            j = random.randint(0, i-1)
            if j < k:
                result[j] = datum
    return result

print R(iter("abcd"), 2)

请注意，算法 R 不为结果提供随机顺序。在给出的示例中，'b' 在结果中永远不会出现在 'a' 之前。

Answer 2

使用O(n)算法R https://en.wikipedia.org/wiki/Reservoir_sampling，从iterable中选择k随机元素：

import itertools
import random

def reservoir_sample(iterable, k):
    it = iter(iterable)
    if not (k > 0):
        raise ValueError("sample size must be positive")

    sample = list(itertools.islice(it, k)) # fill the reservoir
    random.shuffle(sample) # if number of items less then *k* then
                           #   return all items in random order.
    for i, item in enumerate(it, start=k+1):
        j = random.randrange(i) # random [0..i)
        if j < k:
            sample[j] = item # replace item with gradually decreasing probability
    return sample

例子：

>>> reservoir_sample(iter('abcdefghijklmnopqrstuvwxyz'), 5)
['w', 'i', 't', 'b', 'e']

reservoir_sample() 代码来自this answer。

Answer 3

如果您需要具有固定频率的原始迭代器的子集（即，如果生成器生成 10000 个数字，那么您需要“统计上”其中的 100 个，如果它生成 1000000 个数字，您需要其中的 10000 个 - 始终为 1% )，您可以将迭代器包装在一个构造中，以 1% 的概率产生内部循环的结果。

所以我猜你想要一个固定数量的样本，来自未知基数的来源，就像你提到的 Perl 算法一样。

您可以将迭代器包装在一个拥有自己的小内存的构造中，以便跟踪水库，并以降低的概率循环它。

import random

def reservoir(iterator, size):
    n = size
    R = iterator[0:n]
    for e in iterator:
        j = random.randint(0, n-1)
        n = n + 1
        if (j < size):
                R[j] = e
    return R

所以

print reservoir(range(1, 1000), 3)

可能会打印出来

[656, 774, 828]

我已经尝试像上面那样生成一百万轮，并用这个过滤器比较三列的分布（我希望是高斯分布）。

#                get first column and clean it
python file.py | cut -f 1 -d " " | tr -cd "0-9\n" \
    | sort | uniq -c | cut -b1-8 | tr -cd "0-9\n" | sort | uniq -c

虽然（还）不是真正的高斯分布，但它对我来说已经足够好了。

Answer 4

一种可能的方法是围绕迭代器构建一个生成器来选择随机元素：

def random_wrap(iterator, threshold):
    for item in iterator:
        if random.random() < threshold:
            yield item

当您不知道迭代器的长度和可能的大小会令人望而却步时，此方法将很有用。请注意，保证最终列表的大小是有问题的。

一些示例运行：

>>> list(random_wrap(iter('abcdefghijklmnopqrstuvwxyz'), 0.25))
['f', 'h', 'i', 'r', 'w', 'x']

>>> list(random_wrap(iter('abcdefghijklmnopqrstuvwxyz'), 0.25))
['j', 'r', 's', 'u', 'x']

>>> list(random_wrap(iter('abcdefghijklmnopqrstuvwxyz'), 0.25))
['c', 'e', 'h', 'n', 'o', 'r', 'z']

>>> list(random_wrap(iter('abcdefghijklmnopqrstuvwxyz'), 0.25))
['b', 'c', 'e', 'h', 'j', 'p', 'r', 's', 'u', 'v', 'x']

Answer 5

使用itertools.compress() 函数，带有随机选择器函数：

itertools.compress(long_sequence, (random.randint(0, 100) < 10 for x in itertools.repeat(1)))