大數據量樣本隨機採樣-蓄水池算法

最近在個性化推薦系統的優化過程當中遇到一些問題，大體描述以下：目前在咱們的推薦系統中，各個推薦策略召回的item相對較爲固定，這樣就會致使一些問題，用戶在多個推薦場景（若是多個推薦場景下使用了相同的召回策略）、屢次請求時獲得的結果也較爲固定，對流量的利用效率會有所下降；尤爲對於行爲較少的用戶，用來做爲trigger的行爲數據自己就不多，這樣就使得召回item同質化較爲嚴重，使得第一個問題更加明顯。

目前的解決方法是，在推薦策略的召回階段加入必定的隨機機制，使得用戶在多個場景、屢次請求時能後給用戶展現類似但不徹底雷同的結果 。因此問題就轉化爲在N個召回結果（召回結果須要適當地擴大）中隨機抽樣出K個結果，兩個難點：

1. N的值很大時，直接在N個數中取K個數實際是比較慢的，再加上咱們這裏還要求是不重複的採樣，這就致使每次產生的隨機數採樣的結果與以前採樣的某一個結果一致就須要從新進行採樣，這就致使線上計算的性能會受到影響，這個影響隨着N的增長會愈來愈嚴重。因此咱們須要有一種時間複雜度較小的採樣算法，如O(N)的時間複雜度。

2. 對於推薦策略召回的結果，其實每一個item是具備不一樣的權重（類似度）的，因此咱們也能夠利用到這部分信息，即在抽樣時並非等機率採樣，而是帶權重的機率採樣。

 

對於第一個問題，咱們可使用蓄水池算法來解決。首先先看這個問題的簡化版，即從n個數中隨機採樣出1個數。

解法：咱們老是選擇第一個對象，以1/2的機率選擇第二個，以1/3的機率選擇第三個，以此類推，以1/m的機率選擇第m個對象。當該過程結束時，每個對象具備相同的選中機率，即1/n，證實以下。

證實：第m個對象最終被選中的機率P=選擇m的機率*其後面全部對象不被選擇的機率，即

再來看對應的蓄水池抽樣問題，即從n個數中隨機採樣k個數。能夠相似的思路解決。先把讀到的前k個對象放入「水庫」，對於第k+1個對象開始，以k/(k+1)的機率選擇該對象，以k/(k+2)的機率選擇第k+2個對象，以此類推，以k/m的機率選擇第m個對象（m>k）。若是m被選中，則隨機替換水庫中的一個對象。最終每一個對象被選中的機率均爲k/n，證實以下。

證實：第m個對象被選中的機率=選擇m的機率*（其後元素不被選擇的機率+其後元素被選擇的機率*不替換第m個對象的機率），即

實際代碼實現仍是比較簡單的：

 1 List<Map<String, Object>> sampleList = new ArrayList<>();
 2 for (int i=0; i<sampleNum; ++i) {
 3     sampleList.add(rawList.get(i));
 4 }
 5 for (int i=sampleNum; i<rawListSize; ++i) {
 6     int j = r.nextInt(i+1);
 7     if (j < sampleNum) {
 8         sampleList.remove(j);
 9         sampleList.add(rawList.get(i));
10     }
11 }  

再來看看第二個問題，這就涉及到了帶權重的機率抽樣問題了。那有沒有在蓄水池算法基礎上的帶權重機率的抽樣算法呢？固然是有的，想要詳細瞭解的能夠直接看paper《Weighted random sampling with a reservoir》。

首先對於每一個樣本，都具備一個權重Wi，咱們能夠針對這個權重值作一個變換做爲每一個樣本的得分：sampleScore = random(0, 1)^(1/Wi)。而後採樣過程與以前的一致，也是對每一個樣本進行順序讀取。對前k個樣本維護一個最小堆（針對sampleScore排序），而後對於後續的樣本，每次來一個樣本，都將這個新樣本的sampleScore與以前的最小樣本的sampleScore進行比較，若是比最小sampleScore要大，則推出這個最小值，壓入這個新樣本並繼續維護這個最小堆，直到全部樣本都被遍歷過一次。

具體的代碼實現以下：

Comparator<Map<String, Object>> cmp = new Comparator<Map<String, Object>>() {
    public int compare(Map<String, Object> e1, Map<String, Object> e2) {
        return Double.compare((double)e1.get(sampleScoreField), (double)e2.get(sampleScoreField));
        }
};
PriorityQueue<Map<String, Object>> pq = new PriorityQueue<>(sampleNum, cmp);
for (int i=0; i<sampleNum; ++i) {
    Map<String, Object> item = rawList.get(i);
    double sampleScore = Math.pow(r.nextDouble(), 1.0/(0.001+MapUtils.getDoubleValue(item, weightField, 0.0)));
    item.put(sampleScoreField, sampleScore);
    pq.add(item);
}
for (int i=sampleNum; i<rawListSize; ++i) {
    Map<String, Object> item = rawList.get(i);
    double sampleScore = Math.pow(r.nextDouble(), 1.0/(0.001+MapUtils.getDoubleValue(item, weightField, 0.0)));
    item.put(sampleScoreField, sampleScore);
    Map<String, Object> minItem = pq.peek();
    if (sampleScore > (double)minItem.get(sampleScoreField)) {
        pq.remove();
        pq.add(item);
    }
}    

 

以上。

版權聲明：

   本文由笨兔勿應全部，發佈於http://www.cnblogs.com/bentuwuying。若是轉載，請註明出處，在未經做者贊成下將本文用於商業用途，將追究其法律責任。