水塘抽样与阶层固化

V2EX = way to explore

V2EX 是一个关于分享和探索的地方

现在注册

已注册用户请登录

这是一个创建于 2429 天前的主题，其中的信息可能已经有所发展或是发生改变。

简单抽样

简单抽样算法就是从固定的 n 个元素里随机选出 k 个元素，这样每个元素被选的概率都是平等的 k/n。简单抽样是最简单的抽样算法，同样也是使用最为普遍的算法。

简单抽样有个前提就是必须提前知道目标总体的大小 n。我们看看 python 里面的简单抽样算法。

>>> import random
>>> random.sample([1, 2, 3, 4, 5],  3)  # Choose 3 elements
[4, 1, 5]
>>> random.sample([1, 2, 3, 4, 5],  3)  # Choose 3 elements
[5, 3, 1]
>>> random.sample([1, 2, 3, 4, 5],  3)  # Choose 3 elements
[1, 4, 3]

python 内置的简单抽样是无重复抽样，选出来的元素没有重复的。

再细看一下源码实现

def sample(self, population, k):
        n = len(population)
        if not 0 <= k <= n:
            raise ValueError("sample larger than population")
        random = self.random
        _int = int
        result = [None] * k
        setsize = 21
        if k > 5:
            # setsize 的值随着 k 的递增而递增
            # k=1 setsize=4
            # k=[2-5] setsize=16 + 21
            # k=[6-21] setsize=64 + 21
            # k=[22-85] setsize=256 + 21
            # k=[86-341] setsize=1024 + 21
            # ...
            setsize += 4 ** _ceil(_log(k * 3, 4)) # table size for big sets
        if n <= setsize or hasattr(population, "keys"):
            # 如果总数相对太小，就直接使用无放回抽样，因为有放回时重复性较大
            # 每抽出一个元素，原始数组中该元素就空了，然后就被数组尾部的元素替换
            # 数组长度也跟着减 1，砍掉数组尾部的元素（不是物理砍掉，而是缩小随机范围）
            pool = list(population)  # 需要复制母体，如果母体太大，开销就大
            # 如果是 xrange，list()会强迫计算所有元素
            for i in xrange(k): # 无放回抽 k 次
                j = _int(random() * (n-i))  # 缩小随机范围，相当于砍掉数组尾部
                result[i] = pool[j] # 抽到一个拿走
                pool[j] = pool[n-i-1]   # 用数组尾部元素替换被抽走的位置
        else:
            # 如果总数较大，有放回抽样时重复概率不高
            # 尤其适用于 xrange 这种，无须算出所有元素，xrange 是延迟计算的
            # 所以可以使用有放回抽样 + 重复重试法
            try:
                selected = set() # 用于鉴定重复的容器
                selected_add = selected.add
                for i in xrange(k):
                    j = _int(random() * n)
                    while j in selected:  # 重复了，就重试
                        j = _int(random() * n)
                    selected_add(j)
                    result[i] = population[j] # 抽中一个
            except (TypeError, KeyError):
                # 下面不用看了
                if isinstance(population, list):
                    raise
                return self.sample(tuple(population), k)
        return result

代码为 xrange 这种延迟列表进行了优化，无须强制计算出所有元素即可以抽样，比如下面的代码可以瞬间出结果，因为 len(population)和 population[i]都是可以快速计算的。

>>> x=xrange(1,100000000000,8)
>>> random.sample(x, 5)
[683594665, 654156625, 493934665, 580926705, 506506385]

水塘抽样

区别于简单抽样，水塘抽样是一种动态的抽样方法。

抽样的目标总体有一个收集的过程，它是一个动态的过程。简单抽样是要等到这个动态过程彻底冷却下来了，确定了总体的数量之后才会进行一次性抽样。如果目标总体数量又增加了，就必须重新抽样。

动态抽样是渐进式的抽样，它的过程是持续性的。总体在变化，样本也跟着变化，在抽样的过程中是不知道最终会有多少总量的，也就是 n 不确定。

举个例子

原始的部落山寨之间要打战，寨主要选 100 人参战，而这个寨子里总共有 999 人。于是寨主对这 999 人进行了简单的抓阄抽样，选出了 100 个人，准备送上战场，剩下的 899 人开心了。

但是这时突然有人告诉寨主，还有 1 个人刚刚从外地回来，其实总共有 1000 人。

这倒霉的 100 个被选中的人就觉得很不公平了，怎么这刚回来的人就不用筛了呢，这不公平，我们要重新筛选。他们想借此大做文章，这样就有机会不用打战了。

但是另外 899 人不同意，好不容易松弛下来，现在又要来一遍，我们不干。

那该怎么办呢？寨主很快想出了一个办法

可以让这个刚来的小伙伴进行一次抓阄，从 1 ～ 1000 的数中随机拿出一个数，看看它是否小于 100，10%的概率。如果小于 100 的话，很不幸，他就必须上战场。同时之前的 100 个抽中的人中可以随机一个被换下来。这样就可以保证公平，而不会影响之前的 900 个幸运的人，还会给之前抽中的悲惨的 100 人带来一点点小惊喜。

如果接下来又冒出了一个人，那就可以接着上面的方法进行下去，而不会造成混乱。

其实整个过程可以理解为刚开始寨子里只有 100 人，就全部上战场。然后剩下的 900 人一个一个的冒出来，各进行一次随机，随机了 900 次，最终确定了上战场的 100 人选。

越是进行到后面，人选就越固定。因为抓阄的几率是不断变化的。从第 101 人时几率为 100/101 降低到最后一个人的几率为 100/1000。

你会想，这是不是不太公平。其实这是公平的，虽然前面的人被筛进去的几率要高于后面的，但是越早进去的人也就有更大的几率被后面的人替换出来。而最后一个人一旦被筛进去了，就再无翻身的可能。

下面我们使用代码来实现水塘抽样

def reservoir_sampling(items, k):
    # 第一波全部上战场，不用害怕，后面他们会渐渐被新人替换下来
    sample = items[0:k]

    for i in range(k, len(items)):  # 随机 n-k 次
        j = randrange(1, i + 1)  # 抓阄
        if j <= k:
            # 很不幸，中标了，替换掉一个老家伙
            sample[j] = items[i]

    return sample

如果不考虑性能，我们完全可以使用水塘抽样来代替简单抽样。水塘抽样是可以实时看到动态的抽样结果的。