Topk 問題描述

從海量數據中尋找最大(或最小)的 k 個元素，這類問題被稱為 Topk?問題。這個問題無論在實際應用還是面試都會被問到。那我們今天就來看看到底有幾種解決方案，以及各個方案的優劣情況。以下解題思路的前提條件是：從數組array[1, n]中，尋找出最大的 k 個數。

?

全局排序

面對Topk問題，最容易想到的辦法就是排序了。將array里的元素進行排列，便可以獲得最大的 k 個數。此時，Topk問題就轉變成了排序問題，解決Topk問題的時間復雜度變成了排序的時間復雜度。如果使用快排進行排序，那么該問題的時間復雜度就是O(n*lgn)。

?

局部排序

由于是尋找Topk，所以沒有必要對所有的數據都進行排序。雖然快排的表現較好，但是如果使用冒泡或簡單選擇排序，只需要完成k次排序操作就可以解決問題，即如下圖所示。此時的時間復雜度是O(n*k)。注意：局部排序所耗費的時間受 k 值影響。

?

堆

思路：遍歷整個數組，在遍歷的過程中利用小根堆記錄當前的Topk元素。因為小根堆的最小的元素在堆頂，如果下一個元素大于堆頂元素值，那么它就能入選當前的Topk。關于堆的概念和代碼可以看看這篇文章：通俗易懂的講解堆排序(含Gif圖)

例如：從list[?n?] = {58, 32, 73, 20, 31, 95, 46, 77, 22, 67,..., n}中尋找最大的5個數。首先利用前5個元素建立堆，然后再與后續的元素進行對比，不斷的和堆頂元素進行對比。若元素大于堆頂元素，則進行替換，然后調整堆，使其一直保持小根堆的性質。所有元素比對完成后，堆中的元素就是該序列中最大的5個數。

初始建堆：

使用95與堆頂元素對比，若大于堆頂元素，則替換：

替換95后需要調整堆：

重復上述操作，直至所有元素比對完成，堆中的元素就是所求的最大的 5個元素。

?

代碼實現

#include <iostream> #include <time.h> #include <stdlib.h> #include <cstdlib> #include <stdio.h> #include <math.h> #define M 1000000 #define K 50 using namespace std;template <class T> void Print(T a[], int n, int m) {for(int i = n; i < m; i++){cout << "[" << a[i] << "]";}cout <<endl; }template <class T> void Swap(T &a, T &b) {T asd;asd = a;a = b;b = asd; }template <class T> int Partition(T a[], int p, int r) {int i = p, j = r+1;T x = a[p];while(true){while(a[++i] < x && i < r);while(a[--j] > x);if(i >= j)break;Swap(a[i], a[j]);}a[p] = a[j];a[j] = x;return j; }template <class T> void QuickSort(T a[], int p, int r) {if(p < r){int q = Partition(a, p, r);QuickSort(a, p, q-1);QuickSort(a, q+1, r);} }void test(int a[]) {int i,temp;for(i = 0; i < M; ++i){if(a[i] > i)temp = 1;elsetemp = 0;} }void BubbleSort(int a[]) {int i,j,flag,temp;for(i = 0; i < K; ++i){flag = 0;for(j = 0; j < M-i-1; ++j){if(a[j] > a[j+1]){temp = a[j];a[j] = a[j+1];a[j+1] = temp;flag = 1;}}if(flag == 0) break; } }void BubbleSort1(int a[], int n) {int i,j,flag,temp;for(i = 0; i < n; ++i){flag = 0;for(j = 0; j < n-i-1; ++j){if(a[j] > a[j+1]){temp = a[j];a[j] = a[j+1];a[j+1] = temp;flag = 1;}}if(flag == 0) break; } }void heap_ajust_min(int *b, int i, int size) /*a為堆存儲數組，size為堆的大小*/ {int lchild = 2*i; //i的左孩子節點序號 int rchild = 2*i +1; //i的右孩子節點序號 int min = i; /*存放三個頂點中最大的數的下標*/int temp;if(i <= size/2) //假設i是葉節點就不用進行調整 {if(lchild<=size && b[lchild]<b[min]){min = lchild;} if(rchild<=size && b[rchild]<b[min]){min = rchild;}if(min != i){temp = b[i]; /*交換a[i]和a[max]的值*/b[i] = b[min];b[min] = temp;heap_ajust_min(b, min, size); /*被交換的位置曾經是大根堆，如今可能不是大根堆所以須要又一次調整使其成為大根堆結構*/ }} }void build_bheap_min(int *b, int size) /*建立小堆*/ {int i;for(i=size/2; i >= 1; i--) /*非葉節點最大序號值為size/2*/{heap_ajust_min(b, i, size); /*每一個非葉子結點都須要調用這個函數*/ } }int a[M] = {0}; int a1[M] = {0}; int a2[M] = {0}; int a3[M] = {0}; int b[K+1]={0}; int c[K] = {0};int main() {srand(time(0));for(int i = 0; i < M; i++){a[i] = rand()%(M); //設置隨機數組a1[i] = rand()%(M);a2[i] = rand()%(M);a3[i] = rand()%(M);}//方法一clock_t start_time = clock();QuickSort(a, 0, M-1);clock_t end_time = clock();cout<<"快排全排列:"<<static_cast<double>(end_time - start_time)/CLOCKS_PER_SEC*1000<<"ms"<<endl;//方法二clock_t start_time1 = clock();BubbleSort(a1);clock_t end_time1 = clock();cout<<"冒泡排列k次:"<<static_cast<double>(end_time1 - start_time1)/CLOCKS_PER_SEC*1000<<"ms"<<endl;//方法三clock_t start_time2 = clock();for(int i = 0; i < M; ++i){if(i < K){c[i] = a2[i];} else{if(a2[i] > c[0]){c[0] = a2[i];BubbleSort1(c, K);}}}clock_t end_time2 = clock();cout<<"使用冒泡方法記錄Topk:"<<static_cast<double>(end_time2 - start_time2)/CLOCKS_PER_SEC*1000<<"ms"<<endl;//方法四clock_t start_time3 = clock();for(int i = 0; i < M; ++i){if(i <= K){b[i+1] = a3[i];} else{if(a3[i] > b[1]){b[1] = a3[i];build_bheap_min(b, K);}}}clock_t end_time3 = clock();cout<<"使用小根堆記錄Topk:"<<static_cast<double>(end_time3 - start_time3)/CLOCKS_PER_SEC*1000<<"ms"<<endl;return 0; }

?

實驗

在100萬個元素的情況下，分別令 K 的值為10和50。實驗結果如下：

可以發現冒泡排序受 K 的變化而變化，而其他的三種方式較為穩定。四種方法中表現較好的是后兩種，為了比個高低，我設置了1000萬個元素，分別令K值為1000和10000，實驗結果如下：

在擴充了元素和改變 K 值后可發現，利用小根堆的方法性能更好。原因：假設在最壞的情況下，方法三每次完成一次排序，方法四每次都要調整堆。方法三的時間復雜度為O(n*k)，而方法四的時間復雜度為O(n*lg(k))。

總結

以上是生活随笔為你收集整理的Topk 问题详解及代码和数据分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。