給定一個無序的整型數(shù)組arr西雀，找到其中最小的k個數(shù)
該題是互聯(lián)網(wǎng)面試中十分高頻的一道題，如果用普通的排序算法，排序之后自然可以得到最小的k個數(shù)，但時間復雜度高達O(NlogN)鹃答，且普通的排序算法均屬于內(nèi)部排序乎澄，需要一次性將全部數(shù)據(jù)裝入內(nèi)存突硝，對于求解海量數(shù)據(jù)的top k問題是無能為力的。
針對海量數(shù)據(jù)的top k問題置济，這里實現(xiàn)了一種時間復雜度為O(Nlogk)的有效算法：初始時一次性從文件中讀取k個數(shù)據(jù)解恰，并建立一個有k個數(shù)的最大堆，代表目前選出的最小的k個數(shù)浙于。然后從文件中一個一個的讀取剩余數(shù)據(jù)护盈，如果讀取的數(shù)據(jù)比堆頂元素小，則把堆頂元素替換成當前的數(shù)羞酗，然后從堆頂向下重新進行堆調(diào)整腐宋；否則不進行任何操作，繼續(xù)讀取下一個數(shù)據(jù)檀轨。直到文件中的所有數(shù)據(jù)讀取完畢胸竞，堆中的k個數(shù)就是海量數(shù)據(jù)中最小的k個數(shù)（如果是找最大的k個數(shù)，則使用最小堆）参萄。具體過程請參看如下代碼：
public class FindKMinNums {
/**
* 維護一個有k個數(shù)的最大堆卫枝，代表目前選出的最小的k個數(shù)
*
* @param read 實際場景中，read提供的數(shù)據(jù)需要從文件中讀取讹挎，這里為了方便用數(shù)組表示
* @param k
* @return
*/
public static int[] getKMinsByHeap(int[] read, int k) {
if (k < 1 || k > read.length) {
return read;
}
int[] kHeap = new int[k];
for (int i = 0; i < k; i++) { // 初始時一次性從文件中讀取k個數(shù)據(jù)
kHeap[i] = read[i];
}
buildHeap(kHeap, k); // 建堆校赤，時間復雜度O(k)
for (int i = k; i < read.length; i++) { // 從文件中一個一個的讀取剩余數(shù)據(jù)
if (read[i] < kHeap[0]) {
kHeap[0] = read[i];
heapify(kHeap, 0, k); // 從堆頂開始向下進行調(diào)整吆玖，時間復雜度O(logk)
}
}
return kHeap;
}
/**
* 建堆函數(shù)
*
* @param arr
* @param n
*/
public static void buildHeap(int[] arr, int n) {
for (int i = n / 2 - 1; i >= 0; i--) {
heapify(arr, i, n);
}
}
/**
* 從arr[i]向下進行堆調(diào)整
*
* @param arr
* @param i
* @param heapSize
*/
public static void heapify(int[] arr, int i, int heapSize) {
int leftChild = 2 * i + 1;
int rightChild = 2 * i + 2;
int max = i;
if (leftChild < heapSize && arr[leftChild] > arr[max]) {
max = leftChild;
}
if (rightChild < heapSize && arr[rightChild] > arr[max]) {
max = rightChild;
}
if (max != i) {
swap(arr, i, max);
heapify(arr, max, heapSize); // 堆結構發(fā)生了變化，繼續(xù)向下進行堆調(diào)整
}
}
public static void swap(int[] arr, int i, int j) {
int tmp = arr[i];
arr[i] = arr[j];
arr[j] = tmp;
}
public static void printArray(int[] arr) {
for (int i = 0; i <= arr.length; i++) {
System.out.print(arr[i] + " ");
}
System.out.println();
}
public static void main(String[] args) {
int[] arr = {6, 9, 1, 3, 1, 2, 2, 5, 6, 1, 3, 5, 9, 7, 2, 5, 6, 1, 9};
// sorted : { 1, 1, 1, 1, 2, 2, 2, 3, 3, 5, 5, 5, 6, 6, 6, 7, 9, 9, 9 }
printArray(getKMinsByHeap(arr, 10));
}
}
對于從海量數(shù)據(jù)(N)中找出TOP K马篮，這種算法僅需一次性將k個數(shù)裝入內(nèi)存沾乘，其余數(shù)據(jù)從文件一個一個讀即可，所以它是針對海量數(shù)據(jù)TOP K問題最為有效的算法
對于非海量數(shù)據(jù)的情況浑测，還有一種時間復雜度僅為O(N)的經(jīng)典算法 —— BFPRT算法意鲸，該算法于1973年由Blum、Floyd尽爆、Pratt怎顾、Rivest和Tarjan聯(lián)合發(fā)明，其中蘊含的深刻思想改變了世界漱贱。
BFPRT算法解決了這樣一個問題：在時間復雜度O(N)內(nèi)槐雾，從無序的數(shù)組中找到第k小的數(shù)。顯而易見的是幅狮，如果我們找到了第k小的數(shù)募强，那么想求arr中最小的k個數(shù)荆陆，只需再遍歷一遍數(shù)組少梁，把小于第k小的數(shù)都搜集起來，再把不足部分用第k小的數(shù)補全即可鲫忍。
BFPRT算法是如何找到第k小的數(shù)逐抑？以下是BFPRT算法的過程鸠儿，假設BFPRT算法的函數(shù)是int select(int[] arr, int k)，該函數(shù)的功能為在arr中找到第k小的數(shù)厕氨，然后返回該數(shù)进每。select(arr, k)的過程如下：
將arr中的n個元素劃分成 n/5 組，每組5個元素命斧，如果最后的組不夠5個元素田晚，那么最后剩下的元素為一組（n%5 個元素）。時間復雜度O(1)
對每個組進行排序国葬，比如選擇簡單的插入排序贤徒，只針對每個組最多5個元素之間的組內(nèi)排序，組與組之間不排序汇四。時間復雜度 N/5O(1)
找到每個組的中位數(shù)接奈，如果元素個數(shù)為偶數(shù)可以找下中位數(shù)，讓這些中位數(shù)組成一個新的數(shù)組船殉，記為mArr鲫趁。時間復雜度O(N/5)
遞歸調(diào)用
select(mArr, mArr.length / 2)，意義是找到mArr這個數(shù)組的中位數(shù)x利虫，即中位數(shù)的中位數(shù)挨厚。時間復雜度T(N/2)根據(jù)上面得到的x劃分整個arr數(shù)組（partition過程）堡僻，劃分的過程為：在arr中，比x小的都在x左邊疫剃，比x大的都在x右邊钉疫，x在中間。時間復雜度O(N)
-
假設劃分完成后巢价，x在arr中的位置記為i牲阁，關于i與k的相對大小，有如下三種情況：
- 如果 i = k壤躲，說明x為整個數(shù)組中第k小的數(shù)城菊，直接返回。時間復雜度O(1)
- 如果 i < k碉克，說明x處在第k小的數(shù)左邊凌唬，應該在x的右邊繼續(xù)尋找，所以遞歸調(diào)用select函數(shù)漏麦，在右半?yún)^(qū)尋找第k-i小的數(shù)客税。時間復雜度不超過T(7/10N + 6)
- 如果 i > k，說明x處在第k小的數(shù)右邊撕贞，應該在x的左邊繼續(xù)尋找更耻，所以遞歸調(diào)用select函數(shù)，在左半?yún)^(qū)尋找第k小的數(shù)捏膨。時間復雜度同樣不超過T(7/10N + 6)
上述過程的代碼實現(xiàn)如下：
public class FindKMinNums {
/**
* 先用BFPRT算法求出第k小的數(shù)秧均，再遍歷一遍數(shù)組才能求出最小的k個數(shù)，時間復雜度O(N)
* 需要將所有數(shù)據(jù)一次性裝入內(nèi)存脊奋，適用于非海量數(shù)據(jù)的情況
*
* @param arr
* @param k
* @return
*/
public static int[] getKMins(int[] arr, int k) {
if (k < 1 || k > arr.length) {
return arr;
}
int kthMin = getKthMinByBFPRT(arr, k); // 使用BFPRT算法求得第k小的數(shù)熬北，O(N)
int[] kMins = new int[k]; // 下面遍歷一遍數(shù)組疙描，利用第k小的數(shù)找到最小的k個數(shù)诚隙，O(N)
int index = 0;
for (int i = 0; i < arr.length; i++) {
if (arr[i] < kthMin) { // 小于第k小的數(shù)，必然屬于最小的k個數(shù)
kMins[index++] = arr[i];
}
}
while (index < k) {
kMins[index++] = kthMin; // 不足部分用第k小的數(shù)補全
}
return kMins;
}
/**
* 使用BFPRT算法求第k小的數(shù)
*
* @param arr
* @param k
* @return
*/
public static int getKthMinByBFPRT(int[] arr, int k) {
int[] arrCopy = copyArray(arr); // 在得到第k小的數(shù)之后還要遍歷一遍原數(shù)組起胰，所以并不直接操作原數(shù)組
return select(arrCopy, 0, arrCopy.length - 1, k - 1); // 第k小的數(shù)久又，即排好序后下標為k-1的數(shù)
}
/**
* 拷貝數(shù)組
*
* @param arr
* @return
*/
public static int[] copyArray(int[] arr) {
int[] arrCopy = new int[arr.length];
for (int i = 0; i < arrCopy.length; i++) {
arrCopy[i] = arr[i];
}
return arrCopy;
}
/**
* 在數(shù)組arr的下標范圍[begin, end]內(nèi)，找到排序后位于整個arr數(shù)組下標為index的數(shù)
*
* @param arr
* @param begin
* @param end
* @param index
* @return
*/
public static int select(int[] arr, int begin, int end, int index) {
if (begin == end) {
return arr[begin];
}
int pivot = medianOfMedians(arr, begin, end); // 核心操作：中位數(shù)的中位數(shù)作為基準
int[] pivotRange = partition(arr, begin, end, pivot); // 拿到分區(qū)后中區(qū)的范圍
if (index >= pivotRange[0] && index <= pivotRange[1]) { // 命中
return arr[index];
} else if (index < pivotRange[0]) {
return select(arr, begin, pivotRange[0] - 1, index);
} else {
return select(arr, pivotRange[1] + 1, end, index);
}
}
/**
* 選基準
*
* @param arr
* @param begin
* @param end
* @return
*/
public static int medianOfMedians(int[] arr, int begin, int end) {
int num = end - begin + 1;
int offset = num % 5 == 0 ? 0 : 1; // 5個成一組效五，不滿5個的自己成一組
int[] mArr = new int[num / 5 + offset]; // 每組的中位數(shù)取出構成中位數(shù)數(shù)組mArr
for (int i = 0; i < mArr.length; i++) {
int beginI = begin + i * 5;
int endI = beginI + 4;
mArr[i] = getMedian(arr, beginI, Math.min(endI, end));
}
// 求中位數(shù)數(shù)組mArr的中位數(shù)地消，作為基準返回
return select(mArr, 0, mArr.length - 1, mArr.length / 2);
}
/**
* 在數(shù)組arr的下標范圍[begin, end]內(nèi)，找中位數(shù)畏妖，如果元素個數(shù)為偶數(shù)則找下中位數(shù)
*
* @param arr
* @param begin
* @param end
* @return
*/
public static int getMedian(int[] arr, int begin, int end) {
insertionSort(arr, begin, end);
int sum = begin + end;
int mid = (sum / 2) + (sum % 2);
return arr[mid];
}
/**
* 這里僅用于對一組5個數(shù)進行插入排序脉执，時間復雜度O(1)
*
* @param arr
* @param begin
* @param end
*/
public static void insertionSort(int[] arr, int begin, int end) {
for (int i = begin + 1; i <= end; i++) {
int get = arr[i];
int j = i - 1;
while (j >= begin && arr[j] > get) {
arr[j + 1] = arr[j];
j--;
}
arr[j + 1] = get;
}
}
/**
* 優(yōu)化后的快排partition操作
*
* @param arr
* @param begin
* @param end
* @param pivot
* @return 返回劃分后等于基準的元素下標范圍
*/
public static int[] partition(int[] arr, int begin, int end, int pivot) {
int small = begin - 1; // 小區(qū)最后一個元素下標
int big = end + 1; // 大區(qū)第一個元素下標
int cur = begin;
while (cur < big) {
if (arr[cur] < pivot) {
swap(arr, ++small, cur++);
} else if (arr[cur] > pivot) {
swap(arr, --big, cur);
} else {
cur++;
}
}
int[] range = new int[2];
range[0] = small + 1; // 中區(qū)第一個元素下標
range[1] = big - 1; // 中區(qū)最后一個元素下標
return range;
}
public static void swap(int[] arr, int i, int j) {
int tmp = arr[i];
arr[i] = arr[j];
arr[j] = tmp;
}
public static void printArray(int[] arr) {
for (int i = 0; i < arr.length; i++) {
System.out.print(arr[i] + " ");
}
System.out.println();
}
public static void main(String[] args) {
int[] arr = {6, 9, 1, 3, 1, 2, 2, 5, 6, 1, 3, 5, 9, 7, 2, 5, 6, 1, 9};
// sorted : { 1, 1, 1, 1, 2, 2, 2, 3, 3, 5, 5, 5, 6, 6, 6, 7, 9, 9, 9 }
printArray(getKMins(arr, 10));
}
}
關于BFPRT算法為什么在時間復雜度上可以做到穩(wěn)定的O(N)，可以參考《程序員代碼面試指南》P339或《算法導論》9.3節(jié)內(nèi)容戒劫，這里不做證明半夷。
