池塘抽样 Reservoir Sampling

什么是Reservoir Sampling

Reservoir Sampling,水塘抽样算法是随机算法的一种，通常用于选取简单随机样本。

Reservoir Sampling 的用途

对于一个固定样本，样本总数为n，要在其中随机抽取k个样本，我们可以通过在[0,n)中进行随机取数，以保证选取样本的随机性。但是，当n变成一个极大的不固定的数，大到无法将n个样本全部载入到内存中，那么上述通过[0,n)随机数的方式就不能达到期望。需要一种在n不确定情况下，也可以针对全部样本进行随机抽样的算法。Reservoir Sampling可以达到O(n)时间复杂度内与O(k)的空间复杂度。

实现

使用链表结构表示未知大小的样本总数，随机选取k个样本

#1.将[0, k)个样本依次放入reservoir[k]中
#2.遍历I in [k, n)，每次从[0, i]随机一个数r，假设r∈[0, k)，则将reservoir[r]替换为m
class Solution:
    def __init__(self, head: Optional[ListNode]):
        self.head = head

    def getRandom(self, k: int) -> List:
        node, i = self.head, 0
        reservoir = [] 
        for _ in range(k) :
            reservoir.append(node)
            node = node.next
            i += 1
        while node:
            r = randrange(i) 
            if r < k:  
                reservoir[r] = node.val
            i += 1
            node = node.next
        return reservoir

证明

在未知样本streamn大小的前提下，这里对k个样本的随机性进行证明，也就是每个待选样本被选中的概率都是k/n。根据实现方式，可以将证明分为两部分

1. 证明范围在[k, n)的倒数k个数被选取的概率是k/n

对于这个范围内的每一个streami元素，执行的操作都是先从[0,i)随机一个数r，假设r<k,则将reservoirr替换为streami。考虑最后一个样本streamn-1被选中的概率，由于是最后一个样本，因此该样本一旦被选取，后续不会存在再被替换的可能。因此只要随机数小于k，则该样本被选中，即使概率是k/n。而对于倒数第二个样本，即streamn-2,它最终被选取的概率p应该是在遍历到n-2时该样本被选中的概率 乘以 最后一个样本所得到的随机数与上一个样本的随机数不同的概率。streamn-2被选中的概率是k/n-1，streamn-1被选中且不会替换掉streamn-1的概率，等同于streamn-1时的随机数与上一个随机数不同的概率，此时待选随机数有n个，由于上一个随机数是已经发生的事件，该随机数是一个固定的数字，那么在剩余n-1个数中任选一个数必然不会与上一个随机数相同，因此该概率是n-1/n。综上得出，streamn-2最终能被选中的概率p = (k/n-1) * (n-1/n) = k/n。

与以上推到过程类似，我们容易得出[k, n)范围内倒数k个样本，每个样本最终被选取的概率是k/n。

2. 证明[0, k)范围内前k个数，每个数最终被选取的概率是k/n

前个数初始化时就被按序放入reservoirk中，对于每个样本来说，最终被选取的概率，就是在[k, n)过程完成后还没有被替换的概率。以第一个样本为例，在第k+1个样本时不会替换第一个样本的概率，就是在[0, k+1)范围内取随机数不取到1的概率，也就是1 - (1/k+1) = k/k+1，以次类推，第一个样本最终被选取的概率p= k/(k+1) (k+1)/(k+2) ... * (n-1)/n = k/n。同理可得，在[0, k)范围内前k个数，每个数最终被选取的概率是k/n。

参阅

1.geekforgeeks, reservoir sampling

2.leetcode, 链表随机节点

3.wiki, reservoir sampling