top K问题是一个经典的问题。
该问题描述为：输入n个整数，输出其中最小的k个元素，例如，输入 1,2,3,4，5,6,7,8 那么最小的4个元素就是1,2,3,4.
除了这个，top K问题还指：常遇到的一类问题是，在海量数据中找出出现频率最高的前K个数，或者从海量数据中找出最大的前K个数，这类问题通常称为“top K”问题，如：在搜索引擎中，统计搜索最热门的10个查询词；在歌曲库中统计下载率最高的前10首歌等等。
说到top K（第一类）问题，脑袋中经常闪现的两个概念是：快速排序和堆。为什么是这两个概念呢？原因有：
1.快速排序的思路：
给定一个枢轴元素，可以将数组按照这个元素分为两个部分。这个思路对于top K问题有什么作用？答案就是，根据partition的结果（返回的是枢轴的索引），可以轻松得到元素的个数。根据这个数字与K的关系递归划分，最后一定可以得出前面元素个数为k个的划分。
该思路的实现部分可见：http://blog.csdn.net/ohmygirl/article/details/7846544 快速排序求数组的第K个元素。
2.堆。
堆其实是一棵完全二叉树，堆对于两类问题有着很好的解决方案：a.排序问题：由于堆是一棵完全二叉树，所以采用堆堆n元数组进行排序，时间复杂度不会超过O(nlgn),而且只需要几个额外的空间。b.优先级队列。通过插入新元素和调整堆结构来维护堆的性质，每个操作所需要的时间都是O(lgn).
堆的常见实现是采用一个大小为n的数组存储元素，并且0号单元舍弃不用。对堆中的元素按照层次从上到下，从左到右的顺序依次编号。那么对于一个编号为i的元素：
a：如果左孩子存在，那么左孩子的编号为2i
b：如果右孩子存在，那么右孩子的编号为2*i + 1
c：如果有父节点，那么父节点的编号为 i/2
d：节点为叶节点的条件是左孩子且右孩子都为空，为空节点的条件是i<1或者i>n堆的设计对于处理top K问题十分方便。首先设置一个大小为K的堆（如果求最大top K,那么用最小堆，如果求最小top K,那么用最大堆），然后扫描数组。并将数组的每个元素与堆的根比较，符合条件的就插入堆中，同时调整堆使之符合堆的特性，扫描完成后，堆中保留的元素就是最终的结果。说到调整堆，不得不提的是调整的算法，分为两类：
向下调整（shiftdown） 和 向上调整(shiftup)。
以最小堆为例：
向上调整算法 对应的代码如下：
void shiftUp(int *heap,int n){
int i = n;
for(;;){
if(i == 1){
break;
}
int p = i/2;
if(heap[p] <= heap[i]){
break;
}
swap(&heap[p],&heap[i]);
i = p;
}
}向下调整对应的代码如下：
void shiftDown(int * heap,int n){
int i = 1;
for(;;){
int c = 2*i;
if(c > n){
break;
}
if(c+1 <= n){
if(heap[c+1] <= heap[c]){
c++;
}
}
if(heap[i] <= heap[c]){
break;
}
swap(&heap[c],&heap[i]);
i = c;
}
}有了堆的基本操作，top K问题就有了一个基础（当然也可以完全不用堆解决top K问题）。以最小top K问题为例（此时需要建立大小为k的最大堆），top K的求解过程是：扫描原数组，将数组的前K个元素扔到堆中，调整使之保持堆的特性。对于k之后的元素，如果比堆顶元素小，那么替换堆顶元素并调整堆，扫描是数组完成后，堆中保存的元素就是最终的结果。
进一步思考，对于海量数据的处理，top K问题如何实现呢，当然堆算法还是可行的。有没有其他的思路呢。
关于海量数据的处理，推荐july的博客：http://blog.csdn.net/v_july_v/article/details/7382693 如何秒杀99% 的海量数据处理题
另外一个可以参考的博客：http://dongxicheng.org/big-data/select-ten-from-billions/
最近在研究hadoop，所以我的想法是，用hadoop的MapReduce算法实现top K问题，是不是效率更高一些，毕竟，hadoop在海量数据处理，并行计算方面还是蛮有优势的。
MapReduce的思路也很简单。所以编码的话，只需要定义任务类，然后再定义内部的Mapper和Reducer静态类就可以了。
转载一段mapReduce的top K代码（代码未经测试，原文地址：http://www.linuxidc.com/Linux/2012-05/60234.htm）：
package jtlyuan.csdn;
import java.io.IOException;
import org.apache.Hadoop.conf.Configuration;
import org.apache.Hadoop.conf.Configured;
import org.apache.Hadoop.fs.Path;
import org.apache.Hadoop.io.IntWritable;
import org.apache.Hadoop.io.LongWritable;
import org.apache.Hadoop.io.Text;
import org.apache.Hadoop.mapreduce.Job;
import org.apache.Hadoop.mapreduce.Mapper;
import org.apache.Hadoop.mapreduce.Reducer;
import org.apache.Hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.Hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.Hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.Hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.Hadoop.util.Tool;
import org.apache.Hadoop.util.ToolRunner;
//利用MapReduce求最大值海量数据中的K个数
public class TopKNum extends Configured implements Tool {
public static class MapClass extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
public static final int K = 100;
private int[] top = new int[K];
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] str = value.toString().split(",", -2);
try {// 对于非数字字符我们忽略掉
int temp = Integer.parseInt(str[8]);
add(temp);
} catch (NumberFormatException e) {
//
}
}
private void add(int temp) {//实现插入
if(temp>top[0]){
top[0]=temp;
int i=0;
for(;i<99&&temp>top[i+1];i++){
top[i]=top[i+1];
}
top[i]=temp;
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
for(int i=0;i<100;i++){
context.write(new IntWritable(top[i]), new IntWritable(top[i]));
}
}
}
public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
public static final int K = 100;
private int[] top = new int[K];
public void reduce(IntWritable key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
for (IntWritable val : values) {
add(val.get());
}
}
private void add(int temp) {//实现插入
if(temp>top[0]){
top[0]=temp;
int i=0;
for(;i<99&&temp>top[i+1];i++){
top[i]=top[i+1];
}
top[i]=temp;
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
for(int i=0;i<100;i++){
context.write(new IntWritable(top[i]), new IntWritable(top[i]));
}
}
public int run(String[] args) throws Exception {
Configuration conf = getConf();
Job job = new Job(conf, "TopKNum");
job.setJarByClass(TopKNum.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setMapperClass(MapClass.class);
job.setCombinerClass(Reduce.class);
job.setReducerClass(Reduce.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);
return 0;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new TopKNum(), args);
System.exit(res);
}
}
}
/*
* 列举一部分出来：
* 306 306
307 307
309 309
313 313
320 320
346 346
348 348
393 393
394 394
472 472
642 642
706 706
868 868
*/
至此，处理海量数据我们有了新的思路：MapReduce + hadoop
再次感慨，mapReduce真是海量数据处理的神器~
