蓄水池抽樣演算法/水塘取樣演算法
參考：//blog.csdn.net/weixin_43495317/article/details/103943957
//www.cnblogs.com/krcys/p/9121487.html
//zhuanlan.zhihu.com/p/107793995
1、概念介紹
簡介：
水塘抽樣是一種隨機抽樣演算法，它能夠在一個很大的集合中，抽取一部分樣本，並保證每個樣本的選取概率都是相等並隨機的。
特點：
1、選取集合可以非常大，甚至不知道邊界。
2、每個樣本的選取隨機且概率相等。
3、時間複雜度較低，O(n)，節省記憶體。
適用場景：
1、在一些非常大的集合，或者未知大小的集合，不知道邊界的集合，不知道文件總行數的情況下，隨機抽取k個元素。
2、保證每個元素抽取都是均勻隨機的並且概率相等。
3、盡量高效，節省記憶體地抽取
2、常見問題場景
很多大公司的面試題都考察過這個演算法，下面介紹幾個這樣的例題
問題一：
問題：我有一個長度為N的單鏈表,N的值非常大,我不清楚N的確切值.我怎樣能寫一個儘可能高效的演算法來返回鏈表中K個完全隨機的數. 要求： 1、只能遍歷一次鏈表 2、返回的值要均勻隨機，每個的概率要相等
一般的想法就是，我先遍歷一遍鏈表，得到鏈表的總長度 n，再生成一個 [1,n] 之間的隨機數為索引，然後再遍歷鏈表，找到索引對應的節點，不就是一個隨機的節點了嗎？
但題目說了，只能遍歷一次，意味著這種思路不可行。題目還可以再泛化，給一個未知長度的序列，如何在其中隨機地選擇 k 個元素？想要解決這個問題，就需要著名的水塘抽樣演算法了。
問題二：
問題：我們有1T的文本文件存在硬碟中，想隨機抽取幾行，需要保證儘可能少得使用記憶體並且能夠完全隨機。
之前想到的載入到記憶體就不太適合了,但是還可以想到別的辦法,比如每次讀取一行記錄載入到記憶體,記數+1,清空記憶體中行數據,直到最後統計一共多少行,然後根據總行數來計算K個隨機數.如何再取回行對應的數據呢?我們可以再遍歷一遍,一邊遍歷一邊記錄這一行的號碼是不是在k個隨機數中,如果是,則將該行內容保留.
這樣的話遍歷兩次應該可以做到,但是1T數據遍歷兩次的時間消耗是非常高的.
所以還有更好的方案嗎,那就是水塘抽樣演算法.
問題三：
問題：給定一個數據流，數據流長度N很大，且N直到處理完所有數據之前都不可知，如何在只遍歷一遍數據（O(N)）的情況下，能夠隨機選取出這組數據的k個概率相等的均勻抽樣。 要求： 1、僅掃描數據一次 2、空間複雜度為O(N),空間複雜度與整個數據量無關，只與抽樣大小有關。 3、掃描到數據的前n個數據時（n>k），保存當前已掃描數據的k個均勻抽樣。
根據要求，首先體積很大記憶體一次裝不下，不能直接不能直接取N內的k個隨機數，因為N的長度是未知的。此外也不能採用不能先遍歷一遍，然後分塊存儲數據，再隨機選取。最後要求是數據選取絕對隨機的保證。
可以採用水塘抽樣演算法：
3、水塘抽樣演算法原理
問題切入：
給定一個數據流，數據流長度N很大，如何從中隨機選取k個數據，並且要保證每個數據被抽取到的概率相等。
注意：數據流長度N很大，記憶體無法載入全部數據
演算法思路分析
對這個問題，如何做到均勻隨機，概率相等呢？我們以什麼樣的方式去抽取樣本呢？
首先N很大，不知道長度，所以我們無法調用random(N)來取一個1-N之間的隨機數，我們只能將元素一個一個遍歷載入進入記憶體。
比如我們已經遍歷了i個元素，可以從這i個元素中隨機取了一個元素a，那如果現在再給你一個新元素b，你是繼續留著a呢？還是抽取b作為樣本結果呢？以什麼樣的原則來選擇a和b呢？你的選擇能夠保證概率相等嗎？
這裡就用到了我們的蓄水池抽樣演算法。
當k = 1時
首先考慮最簡單的情況，當k = 1，只取一個元素時，如何選取：
假設數據流含有N個數據，要保證每條數據被抽取到的概率相等，那麼每個數被抽取的概率應該為1/N
當遇到第1個數的時候，概率為1，直接保留第一個數為樣本。
當遇到第2個數的時候，概率為1/2，第1個數和第2個數各有1/2的概率被保留。
當遇到第3個數的時候，3被保留的概率為1/3；(之前剩下的數假設1被保留)，2/3的概率 1 被保留，(此時1被保留的總概率為 2/3 * 1/2 = 1/3)
當遇到第4個數的時候，4被保留的概率為1/4，(之前剩下的數假設1被保留)，3/4的概率 1 被保留，(此時1被保留的總概率為 3/4 * 2/3 * 1/2 = 1/4)
…
當遇到第i個數的時候，i被保留的概率為1/i，之前的樣本數保留不被替換的概率也為1/i
所以，依次類推，每個數被保留的概率都是1/N。
通過以上規律可以看出，對於k = 1的情況，數據流中第i個數被保留的概率為1/i，只要採取這種策略，只需要遍歷一遍數據流就可以得到取樣值，並且保證所有數據被選中的概率均為1/N
演算法總結：
水塘取樣演算法總結： 也就是說，在取第i個數據的時候，生成一個0到1的隨機數p，如果p < k / i，替換池中任意一個數替換為第i個數；當p > k / i，繼續保留前面的數。直到數據流結束，返回此k個數。但是為了保證計算準確性，一般是生成一個0到i的隨機數，跟k相比。
4、水塘抽樣演算法實現
演算法圖解：
當K=1時：
如果K=1，即每次只取一個樣本值，那麼逐個遍曆數組，當遇到第i個數時，就以 1 / i 的概率抽取它做樣本值，以（i – 1）/ i 的概率保留原來的樣本值，不替換。
演算法原理如下圖：
當K > 1時：
如果k > 1，即要從數據流N中取K個樣本值，那麼要保證每條數據被抽取到的概率相等，每個數被抽取的概率必然是K / N
演算法實現：對於前K個數，全部保留，作為樣本集合。對於第i （i > k）個數，則以k/i 的概率抽取第i個數，並以1/k 的概率與前面已經選擇的k個數中的任意一個替換。
演算法原理如下圖：
程式碼實現（java）
當K=1時：
/**
* 當k = 1時，即只取1個數
*
* 演算法步驟：
* 1、遍歷集合第1個元素，直接取出，作為樣本數sample
* 2、從集合第2個元素開始，使用下標 i 標記遍歷的第幾個元素。取一個隨機數 rand = random【1，i】
* 如果 rand == 1，則開始替換，即將當前樣本元素，替換為當前遍歷集合N中第i個元素。即sample = N[i]
* 如果 rand > 1，則不替換，保留之前樣本元素。
* 3、繼續往下迭代遍歷，直到結束。最終樣本為sample。
*/
public class ReservoirSampingTest {
public static void main(String[] args) {
// 定義我們要遍歷的集合，原則上是無限大，我們這裡簡化為6個元素。
int[] data = {1, 2, 3, 4, 5, 6};
// 只取一個元素，所以直接定義一個變數，初始化樣本值
int sample = 0;
// 調用抽樣演算法，獲取樣本
sample = reservoirSamping(sample, data);
System.out.println("最終的樣本為：" + sample);
}
/**
* 取樣演算法
* @param sample 返回的取樣的數 （k=1時，這裡不傳也可以）
* @param data 遍歷的大集合
* @return 返回取樣的結果
*/
public static int reservoirSamping(int sample, int[] data) {
// 第一個元素，直接取出
sample = data[1];
for (int i=1; i<data.length; i++) {
// 定義一個隨機數，範圍為[1, i]，注意Random是左閉右開，所以要加1才能把i包括進去
int rand = new Random().nextInt(i) + 1;
// 此時隨機數範圍最小為1，不是0，所以當隨機數等於1時，則替換為當前數
if (rand == 1) {
sample = data[i];
}
}
// 返回取樣的結果
return sample;
}
}
當K>1時，取k個數
跟我們上面的演算法思想是一致的，只是把k=1，替換為了k=k
即概率為P = k / i
/**
* 當k > 1時，取k個數
* 演算法步驟：
* 1、遍歷集合前k個元素，直接取出，作為樣本集合sample[]
* 2、從集合第K+1個元素開始，使用下標 i 標記遍歷的第幾個元素。取一個隨機數 rand = random【1，i】
* 如果 rand < k，則開始替換，即將當前樣本集合sample中第rand個元素，替換為當前遍歷集合N中第i個元素。即 sample[rand] = N[i]
* 如果 rand > k，則不替換，保留之前樣本集合sample中的元素。
* 3、繼續往下迭代遍歷，直到結束。最終獲得k個元素的樣本。
*/
public class ReservoirSampingTest {
public static void main(String[] args) {
// 定義我們要遍歷的集合，原則上是無限大，我們這裡簡化為10個元素。
int[] data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
// 定義我們需要抽樣的樣本集合，假設k=5，抽取5個元素
int[] sample = new int[5];
// 調用抽樣演算法，獲取樣本
sample = reservoirSamping(sample, data);
// 列印最終的樣本集合
for (int s : sample) {
System.out.println("樣本元素：" + s);
}
}
/**
* 水塘抽樣演算法
* @param sample 要採集的樣本集合
* @param data 需要遍歷大集合
* @return 返回抽樣後的樣本集合sample
*/
public static int[] reservoirSamping(int[] sample, int[] data) {
// 假設k為樣本集合的大小
int k = sample.length;
// 前k個元素，直接取出，放到sample中
for (int i=0; i<k; i++) {
sample[i] = data[i];
}
// 從第K+1個元素開始，根據演算法隨機置換sample中的元素
for (int i=k; i<data.length; i++) {
// 取得[1, i]之間的隨機數，注意Random是左閉右開，所以要加1才能把i包括進去
int rand = new Random().nextInt(i) + 1;
// rand小於k，則開始置換，將sample中第rand個元素，替換為data中第i個元素
if (rand < k) {
sample[rand] = data[i];
}
}
// 遍歷一遍，取樣完成，返回樣本集合sample
return sample;
}
}
5、LeetCode中演算法案例
【382. 鏈表隨機節點】：//leetcode-cn.com/problems/linked-list-random-node/
【398. 隨機數索引】：//leetcode-cn.com/problems/random-pick-index/
6、spark分區器（水塘取樣應用案例）
最近在學習spark的分區器的時候，發現一個很有意思的東西水塘取樣演算法，又稱蓄水池抽樣演算法。
spark的分區器主要有三種：
（只有key-value類型的RDD才有分區器，Value類型的RDD分區器的值是None。）
1、HashPartitioner：數據分區規則為 partitionId = Key.hashCode % numPartitions。
特點：採用哈希的方式將同一類型的Key分配到同一個Partition中。
但是當某個key特別多時，容易導致某個分區數非常多，導致各個分區數據不均勻，形成數據傾斜。
2、RangePartitioner：將一定範圍內的數映射到某一個分區內。分區與分區之間數據是有序的，但分區內的元素是不能保證順序的。sortByKey會使用RangePartitioner。
特點：能夠盡量保證每個分區中數據量的均勻。（主要用於數值類型）
3、自定義分區器。
你可以通過partitionBy來指定自己想要用哪種分區器。
關於RangePartitioner：
現在的問題：
在RangePartitioner中，在執行分區之前其實並不知道數據的分布情況，不知道數據的總量，不知道數據的範圍邊界（這個邊界很重要），也不知道數據分布在哪個範圍。
那我們就很難給數據分區，把哪些數據，或者是哪一段範圍的數據分配到對應分區，從而保證各個分區的數據量都比較均勻。因為不知道數據的分布範圍，不好給數據劃分邊界，也不好分區。
這時，就可以用到水塘取樣演算法了。
RangePartitioner在分區的時候，不知道數據總量，數據分布情況，但是它需要知道這些去劃分分區後的邊界。所以它會調用水塘取樣演算法，返回RDD的總取樣數據量，分區ID和，每個分區數據總量，每個分區的取樣數據。
RangePartitioner中是對水塘取樣演算法，進行了變種和加強，有興趣的可以參考：
//www.cnblogs.com/tongxupeng/p/10435976.html
它的主要步驟為：
1、計算總體的數據抽樣大小sampleSize，計算規則是：至少每個分區抽取20個數據或者最多1M的數據量。
2、根據sampleSize和分區數量計算每個分區的數據抽樣樣本數量最大值sampleSizePrePartition
3、根據以上兩個值進行水塘抽樣，返回RDD的總取樣數據量，分區ID和，每個分區數據總量，每個分區的取樣數據。
4、 數據抽樣完成後，需要對不均衡的Partition重新進行抽樣，默認當Partition中包含的數據量大於平均值的三倍時，該Partition是不均衡的。當取樣完成後，利用樣本容量和RDD中包含的數據總量，可以得到整體的一個數據取樣率fraction。利用此取樣率對不均衡的Partition調用sample運算元重新進行抽樣。(此時，各個分區的數據規模已經知道了，就沒必要再水塘取樣了，直接調用sample運算元二次取樣)
5、根據各個分區的key與權重值，調用determineBounds方法計算出步長與邊界
6、計算每個Key所在Partition：當分區範圍長度在128以內，使用順序搜索來確定Key所在的Partition，否則使用二分查找演算法來確定Key所在的Partition。
