问题描述:
有一个不确定规模的字符流,需要采取k个字符,保证字符流中每个字符的采样概率一样,怎么做?
例子:
- 假设数据流只有一个数据,接受数据,发现数据流结束了,则直接返回该数据,该数据返回的概率为1。
- 假设数据流有两个数据,读取了第一个数据,数据流没有结束,又读取第二个数据,发现数据流结束了,因此若保证第一个和第二个数据输出概率一样,则可以生成一个0-1的随机数R,如果R<=0.5,则返回第一个数据,如果R>0.5,则返回第二个数据。
- 假设数据流有三个数据A,B,C,和前面一样要求只输出一个数。现陆续接收到数据A和B,发现数据流没有结束,则必须淘汰一个数据,A和B均以1/2的概率被淘汰,假设淘汰了A,现接受到C,发现数据流结束了,那么可以分析得到,长度为3的数据流每个数据被输出的概率为1/3才能保证正确性,以1/3的概率留下C,以2/3的概率留下B,那么这三个数据被留下并输出的概率分别为:
数据A被留下的概率=(1/2)(2/3)=1/3
数据B被留下的概率=(1/2)(2/3)=1/3
数据C被留下的概率=1/3
总结一下结论:
假设要采用的数量为K,首先构造一个可以容纳K个元素的数组,将序列前K个元素直接放入数组中,然后对于第K+1及以后的元素(假设为第j个)以K/j的概率决定其是否被替换到数组中(保留下来),当遍历完所有元素之后,数组中剩下的元素即采样的样本。
证明:
对于第i个数,i<=k,在K步前直接被选中入数组,保留下来的概率为1,第K+1步,其被替换的概率=(第K+1个样本被选中)*(选择数组中的i替换)=k/(k+1) * 1/k = 1/(k+1),则其不被K+1替换的概率= 1-1/(K+1)=K/(K+1)。以此类推,不被第K+2个元素替换的概率=1-K/(K+2) * 1/K = (K+1)/(K+2),运行到第n步,保留下来的概率 = 被选中的概率 * 不被替换的概率 = 1 * K/(K+1) * (K+1)/(K+2) * ... * (n-1)/n = k/n.
第j个数,j>k,在第j步被选择的概率为 K/j,不被j+1替换的概率 = 1-K/(j+1) * 1/K = j/(j+1),运行到第n步,被保留的概率 = 被选中的概率 * 不被替换的概率 = K/j * j/(j+1) * (j+1)/(j+2) * ... * (n-1)/n = K/n
综上,每一个元素,被保留下来的概率都为 K/n
代码:
import random class ResevoirSample: def __init__(self,size): ''' 选择size个样本到sample[]数组中''' self.size = size self.counter = 0 self.sample = [] def sampling(self,item): self.counter += 1 # 前k个数直接入数组 if len(self.sample)<self.size: self.sample.append(item) return self.sample # i个元素(i>k)开始概率采样 rand = random.randint(1,self.counter) if rand<=self.size: self.sample[rand-1] = item return self.sample