蓄水池取樣演算法
目錄
問題描述分析
取樣問題經常會被遇到，比如：
- 從 100000 份調查報告中抽取 1000 份進行統計。
- 從一本很厚的電話簿中抽取 1000 人進行姓氏統計。
- 從 Google 搜尋 "Ken Thompson"，從中抽取 100 個結果檢視哪些是今年的。
這些都是很基本的採用問題。
既然說到取樣問題，最重要的就是做到公平，也就是保證每個元素被取樣到的概率是相同的。所以可以想到要想實現這樣的演算法，就需要擲骰子，也就是隨機數演算法。（這裡就不具體討論隨機數演算法了，假定我們有了一套很成熟的隨機數演算法了）
對於第一個問題，還是比較簡單，通過演算法生成 [0,100000−1)[0,100000−1) 間的隨機數 1000 個，並且保證不重複即可。再取出對應的元素即可。
但是對於第二和第三個問題，就有些不同了，我們不知道資料的整體規模有多大。可能有人會想到，我可以先對資料進行一次遍歷，計算出資料的數量 N，然後再按照上述的方法進行取樣即可。這當然可以，但是並不好，畢竟這可能需要花上很多時間。也可以嘗試估算資料的規模，但是這樣得到的取樣資料分佈可能並不平均。
演算法過程
終於要講到蓄水池取樣演算法（Reservoir Sampling）了。先說一下演算法的過程：
假設資料序列的規模為 n，需要取樣的數量的為 k。
首先構建一個可容納k 個元素的陣列，將序列的前 k 個元素放入陣列中。
然後從第 k+1 個元素開始，以 k/n 的概率來決定該元素最後是否被留在陣列中（每進來來一個新的元素，陣列中的每個舊元素被替換的概率是相同的）。 當遍歷完所有元素之後，陣列中剩下的元素即為所需採取的樣本。
證明過程
證法1：
對於陣列中第i個數據（i≤k）。在 k 步之前，被選中的概率為 1。當走到第 k+1 步時，被第 k+1 個數據替換的概率 = 第k+1個元素被選中的概率 * 第i個數 被選中替換的概率，即為
對於第j個數據（j>k）。第 j個數據被選中的概率為k/j。不被 第j+1 個元素替換的概率為 。則執行到第 nn步時，被保留的概率 = 被選中的概率 * 不被替換的概率，即條件概率的連乘）：
所以對於其中每個元素，被保留的概率都為 k/n.
證法2：
用數學歸納法證明，每個樣本被選中的概率的都是相等的。
初始情況是，當前只有k個樣本，此時每個樣本都被選中進入池子，也就是k/k=1的概率。
如果我們一共有n個數據點，假設每個資料點被選進入池子的概率相等，都是k/n。
根據數學歸納法的思想，下面我們只要證明如果一共有n+1個數據點，每個資料進入池子中的概率都是k/(n+1)。
對於第n+1個樣本，它進入池子的概率顯然是k/(n+1)。
對於第j個樣本，j≤n，它在前一輪中已經在池子裡的概率為k/n。
下面我們分兩種情況討論：第一種情況，沒被選中，所以
依然留在池子中。這種情況的概率為
第二種情況，被選中但是
沒有被替換。這種情況發生的概率為
兩者相加
所以此時在池子中的概率為
證畢.
程式碼示例
python程式碼示例
import random
'''
名稱：蓄水池取樣
引數：lis-一個list，模擬資料流
k-int型別，表示從資料流中取樣的個數
功能：從不知道什麼時候會結束的資料流中等概率的保留k個數據，本例中使用一個列表模擬了資料流
'''
def Reservoir_Sampling(lis,k):
result=lis[:k] #前k個元素就不模擬資料流了，直接切片讀取了
i=0
while True:
try:
ele=lis[k+i] #嘗試讀取下一個資料，如果讀取失敗，說明資料流結束
except:
return result #資料流結束，返回結果
if random.randint(0,k+i)<k:
result[random.randint(0,k-1)]=ele #如果滿足替換條件進行替換
i+=1
if __name__=="__main__":
lis=list(range(0,10)) #使用一個列表來模擬資料流，在後面會one-by-one讀取資料
#lis=['a','b','c','d','e','f','g','h','i','j']
result=Reservoir_Sampling(lis,5) #從資料流中等概率留下5個數據
print(result) #輸出結果