海量数据Top K问题

Marlowe

大数据

发布于：May 7, 2021

次浏览

2.4k words

9 min

问题引入：10亿个数中找出最大的10000个数（top K问题）

Top K 问题

在大规模数据处理中，经常会遇到的一类问题：在海量数据中找出出现频率最高的前k个数，或者从海量数据中找出最大的前k个数，这类问题通常被称为top K问题。例如，在搜索引擎中，统计搜索最热门的10个查询词；在歌曲库中统计下载最高的前10首歌等。

针对top K类问题，通常比较好的方案是分治+Trie树/hash+小顶堆，即先将数据集按照Hash方法分解成多个小数据集，然后使用Trie树或者Hash统计每个小数据集中的query词频，之后用小顶堆求出每个数据集中出现频率最高的前K个数，最后在所有top K中求出最终的top K。

解决的几种方法

假设场景为：1亿个数中找出最大的1000个数

直接排序

最容易想到的方法是将数据全部排序，然后在排序后的集合中进行查找，最快的排序算法的时间复杂度一般为O（nlogn），如快速排序。但是在32位的机器上，每个float类型占4个字节，1亿个浮点数就要占用400MB的存储空间，对于一些可用内存小于400M的计算机而言，很显然是不能一次将全部数据读入内存进行排序的。其实即使内存能够满足要求（我机器内存都是8GB），该方法也并不高效，因为题目的目的是寻找出最大的1000个数即可，而排序却是将所有的元素都排序了，做了很多的无用功。

局部淘汰法

第二种方法为局部淘汰法，该方法与排序方法类似，用一个容器保存前1000个数，然后将剩余的所有数字——与容器内的最小数字相比，如果所有后续的元素都比容器内的1000个数还小，那么容器内这个1000个数就是最大1000个数。如果某一后续元素比容器内最小数字大，则删掉容器内最小元素，并将该元素插入容器，最后遍历完这1亿个数，得到的结果容器中保存的数即为最终结果了。此时的时间复杂度为O（n+m^2），其中m为容器的大小，即1000。

分治法

第三种方法是分治法，将1亿个数据分成100份，每份100万个数据，找到每份数据中最大的1000个，最后在剩下的100*1000个数据里面找出最大的1000个。如果100万数据选择足够理想，那么可以过滤掉1亿数据里面99%的数据。100万个数据里面查找最大的1000个数据的方法如下：用快速排序的方法。

Hash法

第四种方法是Hash法。如果这1亿个数里面有很多重复的数，先通过Hash法，把这1亿个数字去重复，这样如果重复率很高的话，会减少很大的内存用量，从而缩小运算空间，然后通过分治法或最小堆法查找最大的1000个数。

最小堆

第五种方法采用最小堆。首先读入前1000个数来创建大小为1000的最小堆，建堆的时间复杂度为O（mlogm）（m为数组的大小即为1000），然后遍历后续的数字，并于堆顶（最小）数字进行比较。如果比最小的数小，则继续读取后续数字；如果比堆顶数字大，则替换堆顶元素并重新调整堆为最小堆。整个过程直至1亿个数全部遍历完为止。然后输出当前堆中的所有1000个数字。该算法的时间复杂度为O（nmlogm），空间复杂度是1000（常数）。

分场景方法选择

实际上，最优的解决方案应该是最符合实际设计需求的方案，在时间应用中，可能有足够大的内存，那么直接将数据扔到内存中一次性处理即可，也可能机器有多个核，这样可以采用多线程处理整个数据集。

下面针对不同的应用场景，分析了适合相应应用场景的解决方案。

单机+单核+足够大内存

如果需要查找10亿个查询次（每个占8B）中出现频率最高的10个，考虑到每个查询词占8B，则10亿个查询次所需的内存大约是10^9 * 8B=8GB内存。如果有这么大内存，直接在内存中对查询次进行排序，顺序遍历找出10个出现频率最大的即可。这种方法简单快速，使用。然后，也可以先用HashMap求出每个词出现的频率，然后求出频率最大的10个词。

单机+多核+足够大内存

这时可以直接在内存中使用Hash方法将数据划分成n个partition，每个partition交给一个线程处理，线程的处理逻辑同（1）类似，最后一个线程将结果归并。

该方法存在一个瓶颈会明显影响效率，即数据倾斜。每个线程的处理速度可能不同，快的线程需要等待慢的线程，最终的处理速度取决于慢的线程。而针对此问题，解决的方法是，将数据划分成c×n个partition（c>1），每个线程处理完当前partition后主动取下一个partition继续处理，知道所有数据处理完毕，最后由一个线程进行归并。

单机+单核+受限内存

这种情况下，需要将原数据文件切割成一个一个小文件，如次啊用hash(x)%M，将原文件中的数据切割成M小文件，如果小文件仍大于内存大小，继续采用Hash的方法对数据文件进行分割，知道每个小文件小于内存大小，这样每个文件可放到内存中处理。采用（1）的方法依次处理每个小文件。

多机+受限内存

这种情况，为了合理利用多台机器的资源，可将数据分发到多台机器上，每台机器采用（3）中的策略解决本地的数据。可采用hash+socket方法进行数据分发。

重点讲下最小堆算法

在几千亿个数据中如何获取10000个最大的数？

一个复杂度比较低的算法就是利用最小堆算法，它的思想就是：先建立一个容量为K的最小堆，然后遍历这几千亿个数，如果对于遍历到的数大于最小堆的根节点，那么这个数入堆，并且调整最小堆的结构，遍历完成以后，最小堆的数字就是这几千亿个数中最大的K个数了。

先来介绍一下最小堆：最小堆（小根堆）是一种数据结构，它首先是一颗完全二叉树，并且，它所有父节点的值小于或等于两个子节点的值。最小堆的存储结构（物理结构）实际上是一个数组。

因为它是一个完全二叉树，对于下标小于数组.length/2 - 1 时有叶子节点，对于下标为i（基0），其左节点下标为2i + 1，右节点下标为2i + 2。

最小堆如图所示，对于每个非叶子节点的数值，一定不大于孩子节点的数值。这样可用含有K个节点的最小堆来保存K个目前的最大值(当然根节点是其中的最小数值)。

每次有数据输入的时候可以先与根节点比较。若不大于根节点，则舍弃；否则用新数值替换根节点数值。并进行最小堆的调整。

代码实现：创建堆的复杂度是O(N)，调整最小堆的时间复杂度为O(logK)，因此Top K算法(问题)时间复杂度为O(NlogK)。

class TopK {
   //创建堆
   int[] createHeap(int a[], int k) {
       int[] result = new int[k];
       for (int i = 0; i < k; i++) {
           result[i] = a[i];
       }
       //完全二叉树的数组表示中，下标小于等于result.length / 2 - 1才有子节点
       for (int i = result.length / 2 - 1;i >= 0;i--){
           heapify(i,result);
       }
       return result;
   }

   void heapify(int i,int[] result){
       int left = 2 * i + 1;
       int right = 2 * i + 2;

       int smallest = i;
       if (left < result.length && result[left] < result[i]){
           smallest = left;
       }
       if (right < result.length && result[right] < result[smallest]){
           smallest = right;
       }
       if (smallest == i){
           return;
       }
       else {
           int temp = result[i];
           result[i] = result[smallest];
           result[smallest] = temp;
       }
       heapify(smallest,result);
   }

   //调整堆
   void filterDown(int a[], int value) {
       a[0] = value;
       int parent = 0;

       while(parent < a.length){
           int left = 2*parent+1;
           int right = 2*parent+2;
           int smallest = parent;
           if(left < a.length && a[parent] > a[left]){
               smallest = left;
           }
           if(right < a.length && a[smallest] > a[right]){
               smallest = right;
           }
           if(smallest == parent){
               break;
           }else{
               int temp = a[parent];
               a[parent] = a[smallest];
               a[smallest] = temp;
               parent = smallest;
           }
       }
   }

    //遍历数组，并且调整堆
   int[] findTopKByHeap(int input[], int k) {
       int heap[] = this.createHeap(input, k);
       for(int i=k;i<input.length;i++){
           if(input[i]>heap[0]){
               this.filterDown(heap, input[i]);
           }

       }
       return heap;

   }

   public static void main(String[] args) {
       int a[] = { 100,101,5,4,88,89,845,45,8,4,5,8,452,1,5,8,4,5,8,4,588,44444,88888,777777,100000};
       int result[] = new TopK().findTopKByHeap(a, 5);
       for (int temp : result) {
           System.out.println(temp);
       }
   }
}