尋找最優的 TopN 演算法

1 概要
在大量的資料記錄中，依據某可排序的記錄屬性（一般為數字型別），找出最大的前 N 個記錄，稱為
TopN 問題。這是一個常常遇到的問題，也是一個比較簡單的演算法問題，卻很少能有人能寫出最優化的
topn 演算法。本文對常見的 TopN 演算法，進行分析比較，最後給出最優的 TopN 演算法：基於小根堆的篩選
法.
2 問題定義
為了方便。我們把問題具體化：要求使用 C 語言來實現函式，在一個大小為 m 的，無序的整數陣列中選
取最大的 n 個整數。即，問題定義為一個函式的實現:
int * topn(int *pdata, int m,int *ptop, int n);
其中，pdata 指向儲存大量整數的記憶體，整數的個數為 m，要求函式把前 n 個最大的整數儲存到 ptop 指
向的記憶體中,顯然滿足 m≥n≥0 。 同時假定 n<=m/2， 因為當 n 如果超過半數，可以問題轉換為排
除 n−m/2個最小的整數----這種轉換減少了問題規模。並且，已經實現選擇 n 個最值（最大或最小）
的演算法的情況下，排除 n 個最值的演算法可以仿照實現。這裡不再費時間敘述。
我們雖然假設的資料型別為整數，其他任意的資料型別，只要其資料是可以排序的，都可以根據這裡的
演算法，寫出對應的演算法實現。同樣，我們雖然假定輸入資料的方式為陣列，實際輸入的方式可以是任意
其他方式（如檔案，資料庫等）。
3 演算法介紹
3.1 演算法 1 完整排序法
對 pdata 中的資料，進行內部排序，然後選取前 n 個整數放到 ptop 中。如果使用較高效率的內部排序
演算法（如：快速排序，堆排序），複雜度為 m∗log2
m .
使用這種演算法，結果資料是從大到小排序的。
3.2 演算法 2 基於有序陣列的篩選法
保持 ptop 中的結果陣列有序，即滿足 ptop[0]≥ptop[1]≥...≥ptop[n−1] ，遍歷 pdata 中的數
據，如果發現當前整數大於 ptop[n-1]就把資料插入到合適位置，以保持 ptop 中的仍然有序。演算法復
雜度為. m∗n
使用這種演算法，結果資料是從大到小排序的。顯然，只從理論上的複雜度考慮，就可以看出來：當 n 較小的時候，演算法 2 有較高的效率。但當 n 較大
的時候，演算法 1 會有較高的效率。
仔細分析演算法 2, 保持 ptop 中資料有序的目的，就是每次能找到最小的值，在有序陣列中插入一個新元
素，最壞需要 n 次操作才能保持陣列有序。其實要快速找到最小的值，使用小根堆更合適，因為中堆中
每次插入一個元素，最壞只需要 log2
n 次操作，根據這個思路產生下面的演算法 3。
3.3 演算法 3 基於小根堆的篩選法。
使用大小為 n 的小根堆儲存結果集，最小項為堆的根節點 ptop[0],遍歷 pdata 中，如果當前資料大於
ptop[0],並把資料替換根節點，並對根節點 SiftDown 操作，保持結果集仍然為小根堆。演算法複雜度，
只有 m∗log2
n .- 更多關於小根堆的介紹，請看後面。
顯然有，
m∗log2 nm∗log2 m , m∗log2 nm∗n
所以，演算法 3 的複雜度最低。
這種演算法，生成的結果資料是沒有排序的。因為結果已經儲存最小根堆種，如果需要排序，只需要完成
常規堆排序流程的後半部分，排序需要的複雜度只有 n。
4 實際測試結果
測試環境：
OS: Debain Linux kernel 2.6.26
C Complier: gcc 4.3.4
MEM: 768m
CPU: Intel(R) Pentium(R) 4 CPU 2.40GHz

5 演算法分析
1. 在任何情況下，演算法 3 都是最優的。但演算法 3 的實現較複雜。
2. 如果不想實現複雜的演算法 3，如果 n 相對於 m 很小，可以選擇演算法 2
3. 如果不想實現複雜的演算法 3, 如果 n 比較大，例如：接近了 m/2, 可以選擇演算法 1
4. 如果不要求結果排序，演算法 3 更有優勢。
5. 演算法 1 和演算法 2，結果自然就排號順序了。是否要求排序與其時間花費無關。
根據理論複雜度和實際的測試結果，更有下面的結論。
當 m 值固定，n 的變化區間為[1,m/2],三種演算法消耗時間隨 n 變化的曲線，如下面的圖示意。相應的，當 n 固定，m 的變化區間為[2*n, ∞ ),三種演算法消耗演算法隨 m 變化的曲線，如下面的圖示意。

6 演算法實現
6.1 演算法 1 實現
使用標準 c 語言庫支援的 qsort，直接就可以實現。
//用於比較操作的函式
static int cmpint(const void *p1, const void *p2)
{
return *((int *)p2) - *((int *)p1);
}
//演算法實現
int * topn1(int *pdata, int m,int *ptop, int n)
{
qsort(pdata, m, sizeof(int),cmpint);
memcpy(ptop, pdata, n*sizeof(int));
return ptop;
}
注意：如果自己實現 qsort 演算法，會有一定效率的提高。但程式碼級的優化，這裡不會對效率產生本質的
影響。6.2 演算法 2 實現
比較簡單，原始碼如下
int *topn2(int *pdata,int m,int *ptop,int n)
{
int i;
for(i=0;i<m;i++)
{
//判斷當前結果集的大小（陣列的長度）
int t = (i+1<n) ? (i+1):n;
if(i>=n && pdata[i] <= ptop[n-1])
continue;
int j=t-1;
for(;j>=1 && pdata[i]>ptop[j-1];j--)
ptop[j] = ptop[j-1];
ptop[j] = pdata[i];
}
return ptop;
}
注意：如果使用監視哨等優化手段，可以減少迴圈的判斷條件。但這種程式碼級別的優化，這裡不會對效
率產生本質的影響。
6.3 演算法 3 實現
6.3.1 小根堆概念定義
小根堆精確定義：陣列 x[1..n](注意，這裡假定陣列的第一個元素的索引為 1，是為了下面描述方便),
如果滿足：
∀i∋[2,n], x[i/2]≤x [i]
就說明，這個陣列就是小根堆
為了下面描述方便，我們定義對陣列 x[1..n]的子陣列 x[u..v]，（ 1≤u≤v≤n ）,如果滿足：
∀i∋[2l, u],x[i/2]≤x [i]
就說明：x 整個陣列的[u..v]部分已經是小根堆了。
6.3.2 我們需要的堆操作
//如果[2..n]已經是堆了，-除了第一個元素，其他部分已經是小根堆。
//通過下面的函式把第一個元素，往後交換，把整個陣列變成堆。
void siftdown_head(int *pheap,int n)
{
pheap --; //這樣作是讓第一個元素的索引變成 1, pheap[0]永遠不會被訪問到int t = pheap[1];
int i =1;
int c = i*2;
while(c<=n)
{
if(c<n && pheap[c]>pheap[c+1])
c++;
if(pheap[c] < t)
{
pheap[i] = pheap[c];
i = c;
c = 2*i;
}
else
break;
}
pheap[i] = t;
}
//如果除了最後一個元素外，其他部分[1..n-1]已經是堆了，
//通過下面的函式，把最後一個元素往前交換，把整個陣列調整成堆
void siftup_rear(int *pheap,int n)
{ 
pheap --;
int t = pheap[n];
int i;
for(i=n; i>1;)
{
int p = i/2;
if( pheap[p] > t)
pheap[i] = pheap[p];
else
break;
i=p;
}
pheap[i] =t;
}
更多關於堆的概念，可以在網上找到。最常見的應用是使用堆來實現高效率的優先順序佇列（如 c++的標
準模板庫）。在本文應用場景中，也可以把儲存 topn 結果的陣列看成是優先順序佇列，因為我們總是選擇
數值最小的值，進行出隊操作（數值越小，優選級越大）。最後，佇列中剩餘的，自然就是數值較大的
topn 結果。
6.3.3 演算法實現
int * topn3(int *pdata, int m,int *ptop, int n){
ptop[0] = pdata[0];
int i;
for(i=1; i<n;i++)
{
ptop[i] = pdata[i];
siftup_rear(ptop,i+1);
}
for(i=n;i<m;i++)
{
register t = pdata[i];
if(ptop[0] >= t)
continue;
ptop[0] = t;
siftdown_head(ptop,n);
;
}
//如果不需要把結果排序，下面的這段迴圈可以乎略掉
for(i=n-1; i>=1;i--)
{
int t = ptop[i];
ptop[i] = ptop[0];
ptop[0] = t;
siftdown_head(ptop,i);
}
return ptop;
}
其實，在本文問題定義下，也可以直接使用 c++標準庫的模板種的優先順序佇列，來實現 topn 演算法。

轉自mail:cuichaox @gmail.com
轉自blog: http://cuichaox.cublog.cn