怎樣對10億個數字快速去重？——淺析位圖數據結構及其應用

最近有個朋友問我一個算法題——

給你幾億個QQ號，怎樣快速去除重復的QQ號？

可以作如下假定：

QQ號數字范圍從0到十億，即[0, 1000000000)，且最多給你10億個QQ號，這些QQ號放在1或多個文本文件中，格式是每行一個QQ號。

 

請讀者先獨立思考一下該怎樣解決。

 

————————————————————————————————————————————————————————

 

其實在一年前碰過類似的問題，當時的解決方案：借助hash算法思想，把一個大文件哈希分割到多個小文件中，而哈希沖突的數字

一定會在同一個小文件中，從而保證了子問題的獨立性，然后就可以單獨對小文件通過快速排序來去重。

這樣就通過分而治之解決了幾G數據進行內排序的問題。

雖然哈希分割文件是O(n)的時間復雜度，但總的效率仍是依從快速排序的時間復雜度O(n*logn)。

另外，分而治之有個好處就是借助子問題的獨立性可以利用多核來做並行處理，甚至做分布式處理。

 

后來小菜在《編程珠璣》中看到了位圖這個數據結構可以很方便地處理此類問題，時間復雜度可以達到了O(n)

那怎么實現這個數據結構呢？

位圖的原理類似我們常用的標記數組map[]/vis[]，比如map[i] = 1表示把第i個元素標記為1，按照這種思想來去重是很簡單的。

現在假定QQ號數字范圍是[0, 10億)，則要申請10億個char元素用來做標記，那么進程就需要1G的運行內存。

那如果數字范圍增大到100億，一般的計算機可能就吃不消了。

 

位圖數據結構只需要1/8的空間，節省7/8的內存是非常可觀的。

因為標記只有1和0兩個值，所以可以只用一個比特位來做標記。

設有char類型數x，1字節包括8個位，我們可以申請char bit_map[10億/8+1]的空間，就足以給范圍在[0,10億)的數字去重了。

選擇char類型而不是int等其它類型是考慮到，C標准規定任何實現都要保證char類型占1個字節。

+1，是考慮到C整型除法向下取整的特點，比如100/8結果為12，這樣訪問編號>=96的比特位（設從0開始編號），就會發生數組越界。

 

我們知道位圖的數據結構就是一個數組，而位圖的操作（算法）基本依賴於下面3個元操作

set_bit(char x, int n); //將x的第n位置1，可以通過x |= (1 << n)來實現

clr_bit(char x, int n); //將x的第n位清0，可以通過x &= ~(1 << n)來實現

get_bit(char x, int n); //取出x的第n位的值，可以通過(x >> n) & 1來實現

 

有了上面3個元操作，位圖的具體操作就簡單了——

比如，要對數字int x = 1848105做標記，就可以調用set_bit(bit_map[x/8], x%8);

除法看做求“組編號”，x/8即是 以8個位為一個小組，分組到編號為idx = x/8的bit_map元素中，然后在組內偏移lft = x%8個比特位。

考慮到這些操作是非常頻繁的，所以把上述三個方法改寫成宏減少函數調用的開銷，並且把x/8改為x<<3，x%8改為x&7。

 

經過上面的分析，寫代碼就很不難了——

/*
 *CopyRight (C) Zhang Haiba
 *File: 1billon_remove_duplicate_not_sort.c
 *Date: 2014.03.11
 */
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#define MAP_LEN (1000000000/8 + 1)
#define BUF_SIZE 10
#define SET_BIT(x, n) ( (x) |= (1 << (n)) )
#define GET_BIT(x, n) ( ((x)>>(n)) & 1 )

char bit_map[MAP_LEN];

int main(int argc, const char *argv[])
{
    FILE *ifp, *ofp;
    int idx, lft, x;
    char buf[BUF_SIZE]; //cut if number length > BUF_SIZE ex. range[0, 1000000000) then BUF_SIZE=10

    if (argc == 1) {
        fprintf(stderr, "usage: %s inputfile1 inputfile2 ...\n", argv[0]);
        exit(1);
    } else {
        ofp = fopen("./output.txt", "w");
        for (idx = 1; idx <= argc; ++idx) {
            if ( (ifp = fopen(argv[idx], "r")) == NULL ) {
                fprintf(stderr, "%s: can not open %s\n", argv[0], argv[idx]);
                exit(1);
            }
            printf("processing the %dth file...\n", idx);
            while ( fgets(buf, sizeof buf, ifp) != NULL ) {
                sscanf(buf, "%d", &x);
                idx = x >> 3;
                lft = x & 7;
                if (GET_BIT(bit_map[idx], lft) == 0) {
                    bit_map[idx] = SET_BIT(bit_map[idx], lft);
                    fprintf(ofp, "%d\n", x);
                }
            }
            fclose(ifp);
        }
        fclose(ofp);
    }
    return 0;
}

 

【測試用例1：】

ZhangHaiba-MacBook-Pro:KandR apple$ time ./a.out  input2.txt 
processing the 1th file...

real    0m0.028s
user    0m0.001s
sys    0m0.002s

輸入輸出文件對比：

由於實現中故意使用了fgets()，可以防止輸入文本中長度不合法的數據

對於長度超過限制，則進行截斷處理（見上圖左邊第一行），同時可以達到濾空的效果。

 

【測試用例2：】

我們可以寫一個小程序生成N個范圍[0, 10億）的數字，也就是最大的數是包含9個9的999999999。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>

#define MAX 1000000000

int main(void)
{
    srand((unsigned)time(NULL));

    fprintf(stderr, "this prog output N random numbers to stdout.\nPlease enter the value of N:\n");
    int n, i;
    scanf("%d", &n);
    for (i = 0; i < n; ++i)
        printf("%d\n", rand()%MAX);
    return 0;
}

通過這個程序生成1億個隨機數並重定向輸出到input1.txt，則這個文本文件大概有970Mb，然后執行測試

ZhangHaiba-MacBook-Pro:KandR apple$ time ./a.out  input1.txt
processing the 1th file...

real    1m12.263s
user    1m0.716s
sys    0m2.685s

耗時1分12秒，速度飛快！

 

如果需要輸出的文本內容是有序的，稍作修改即可——

/*
 *CopyRight (C) Zhang Haiba
 *File: 1billon_remove_duplicate_sort.c
 *Date: 2014.03.11
 */

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#define MAP_LEN (1000000000/8 +1)
#define BUF_SIZE 10
#define CLR_BIT(x, n) ( (x) &= ~(1 << (n)) )
#define SET_BIT(x, n) ( (x) |= (1 << (n)) )
#define GET_BIT(x, n) ( ((x)>>(n)) & 1 )

char bit_map[MAP_LEN];

int main(int argc, const char *argv[])
{
    FILE *fp;
    int idx, lft, x;

    char buf[BUF_SIZE]; //cut if number length > BUF_SIZE ex. range[0, 1000000000) then BUF_SIZE=10
    if (argc == 1) {
        fprintf(stderr, "usage: %s inputfile1 inputfile2 ...\n", argv[0]);
        exit(1);
    } else {
        //memset(bit_map, 0, sizeof bit_mape);
        for (idx = 1; idx <= argc; ++idx) {
            if ( (fp = fopen(argv[idx], "r")) == NULL ) {
                fprintf(stderr, "%s: can not open %s\n", argv[0], argv[idx]);
                exit(1);
            }
            printf("processing the %dth file...\n", idx);
            while ( fgets(buf, sizeof buf, fp) != NULL ) {
                sscanf(buf, "%d", &x);
                idx = x >> 3;
                lft = x & 7;
                bit_map[idx] = SET_BIT(bit_map[idx], lft);
            }
            fclose(fp);
        }

        fp = fopen("./output.txt", "w");
        printf("output to file: output.txt...\n");
        for (idx = 0; idx < MAP_LEN; ++idx) {
            for (lft = 0; lft < 8; ++lft)
                if (GET_BIT(bit_map[idx], lft) == 1)
                    fprintf(fp, "%d\n", (idx<<3)+lft);
        }
        fclose(fp);
    }
    return 0;
}

實際測試發現，對於很小的輸入文本（例如空文本），這種方法也需要3~4秒的本機執行時間用於遍歷輸出。

但對於上面將近1G的輸入文本文件，測試時間與不排序的實現方案相差無幾，甚至略快一點。

 

@Author: 張海拔

@Update: 2014-3-11

@Link: http://www.cnblogs.com/zhanghaiba/p/3594559.html