10000亿数据寻找 最大 或者最小 n个 数 各种算法比较

寻找最优的 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++)

{

//判断当
c108
前结果集的大小（数组的长度）

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
;

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