Topk 问题详解及代码和数据分析

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))。