海量數據處理

基本方法

---

一、Hash法

參考：

• 常見hash算法的原理
• 常見的Hash算法
• General Purpose Hash Function Algorithms

---

二、Bit-map法

利用Bit-map方法解決海量數據的查重和去重問題。

/*已知某個文件內包含一些電話號碼，每一個號碼爲8位數字，統計不一樣號碼個數*/
 
#include <iostream>
#include <time.h>
#include <cstdlib>
using namespace std;
 
#define minNumber 10000000
#define maxNumber 99999999
int N = (maxNumber-minNumber+1); //90000000
#define ARRNUMBER 100            //號碼數組長度
#define BITS_PER_WORD 32         //一個整型4個字節即32位
#define WORD_OFFSET(b) (b/BITS_PER_WORD)
#define BIT_OFFSET(b) (b%BITS_PER_WORD)
 
/*置1*/
void setBit(int *words,int n)
{
    n-=minNumber;
    words[WORD_OFFSET(n)] |= (1<<BIT_OFFSET(n));
}
 
/*清零*/
void clearBit(int *words,int n)
{
    words[WORD_OFFSET(n)] &= ~(1<<BIT_OFFSET(n));
}
 
int getBit(int *words,int n)
{
    return words[WORD_OFFSET(n)]&(1<<BIT_OFFSET(n));
}
 
int main()
{
    int i,j;
    int number = 0;
    int arr[ARRNUMBER];
    int *words = new int[1+N/BITS_PER_WORD];
    if(words == NULL)
    {
        cout<<"new error"<<endl;
        exit(0);
    }
    for(i=0;i<N;i++)
    {
        clearBit(words,i);
    }
 
    srand(time(NULL));//設置種子
 
    for(j=0;j<ARRNUMBER;j++)
    {
        arr[j]=rand()%N;
        arr[j]+=minNumber;
        cout<<arr[j]<<"\t";
    }
    for(j=0;j<ARRNUMBER;j++)
    {
        setBit(words,arr[j]);
    }
    for(i=0;i<N;i++)
    {
        if(getBit(words,i))
        {
            //cout<<i+minNumber<<"\t";
            number++;
        }
    }
 
    cout<<"總個數爲:"<<number<<"\n";
    delete[] words;
    words = NULL;
    return 0;
}

使用C++的bitset實現起來更加簡單。

---

三、Bloom filter法

下面是一個簡單的Bloom Filter的實現：

package com.flyoung;
 
import java.util.BitSet;
//傳統的Bloom filter 不支持從集合中刪除成員。
//Counting Bloom filter因爲採用了計數，所以支持remove操做。
//基於BitSet來實現，性能上可能存在問題
public class SimpleBloomFilter {
    //DEFAULT_SIZE爲2的25次方
    private static final int DEFAULT_SIZE = 2 << 24;
    /* 不一樣哈希函數的種子，通常應取質數,seeds數據共有7個值，則表明採用7種不一樣的HASH算法 */
    private static final int[] seeds = new int[] { 5, 7, 11, 13, 31, 37, 61 };
    //BitSet實際是由「二進制位」構成的一個Vector。假如但願高效率地保存大量「開－關」信息，就應使用BitSet.
    //BitSet的最小長度是一個長整數（Long）的長度：64位
    private BitSet bits = new BitSet(DEFAULT_SIZE);
    /* 哈希函數對象 */
    private SimpleHash[] func = new SimpleHash[seeds.length];
 
    public static void main(String[] args) {
       String value = "flyoung2008@gmail.com";
       //定義一個filter，定義的時候會調用構造函數，即初始化七個hash函數對象所須要的信息。
       SimpleBloomFilter filter = new SimpleBloomFilter();
       //判斷是否包含在裏面。由於沒有調用add方法，因此確定是返回false
       System.out.println(filter.contains(value));
       filter.add(value);
       System.out.println(filter.contains(value));
    }
    //構造函數
    public SimpleBloomFilter() {
       for (int i = 0; i < seeds.length; i++) {
           //給出全部的hash值，共計seeds.length個hash值。共7位。
           //經過調用SimpleHash.hash(),能夠獲得根據7種hash函數計算得出的hash值。
           //傳入DEFAULT_SIZE(最終字符串的長度），seeds[i](一個指定的質數)便可獲得須要的那個hash值的位置。
           func[i] = new SimpleHash(DEFAULT_SIZE, seeds[i]);
       }
    }
 
    // 將字符串標記到bits中，即設置字符串的7個hash值函數爲1
    public void add(String value) {
       for (SimpleHash f : func) {
           bits.set(f.hash(value), true);
       }
    }
 
    //判斷字符串是否已經被bits標記
    public boolean contains(String value) {
       //確保傳入的不是空值
       if (value == null) {
           return false;
       }
       boolean ret = true;
       //計算7種hash算法下各自對應的hash值，並判斷
       for (SimpleHash f : func) {
           //&&是boolen運算符，只要有一個爲0，則爲0。即須要全部的位都爲1，才表明包含在裏面。
           //f.hash(value)返回hash對應的位數值
           //bits.get函數返回bitset中對應position的值。即返回hash值是否爲0或1。
           ret = ret && bits.get(f.hash(value));
       }
       return ret;
    }
    /* 哈希函數類 */
    public static class SimpleHash {
       //cap爲DEFAULT_SIZE的值，即用於結果的最大的字符串長度。
       //seed爲計算hash值的一個給定key，具體對應上面定義的seeds數組
       private int cap;
       private int seed;
 
       public SimpleHash(int cap, int seed) {
           this.cap = cap;
           this.seed = seed;
       }
 
       //計算hash值的具體算法,hash函數，採用簡單的加權和hash
       public int hash(String value) {
           //int的範圍最大是2的31次方減1，或超過值則用負數來表示
           int result = 0;
           int len = value.length();
           for (int i = 0; i < len; i++) {
              //數字和字符串相加，字符串轉換成爲ASCII碼
              result = seed * result + value.charAt(i);
              //System.out.println(result+"--"+seed+"*"+result+"+"+value.charAt(i));
           }
           //sSystem.out.println("result="+result+";"+((cap - 1) & result));
           //System.out.println(414356308*61+'h');  //執行此運算結果爲負數，爲何？
           //&是java中的位邏輯運算，用於過濾負數（負數與進算轉換成反碼進行）。
           return (cap - 1) & result;
       }
    }  
}

參考：

• BloomFilter–大規模數據處理利器(解決空查問題)
• 深刻理解Bloom Filter
• [大數據量]布隆過濾器(Bloom Filter)適用類型以及具體示例

---

四、數據庫優化法

---

五、倒排索引法

---

六、外排序法

---

七、Trie樹

Trie樹的典型應用是用於統計和排序大量的字符串，常常被搜索引擎系統用於文本詞頻統計。
下面是一個利用Trie統計某個單詞出現的頻數的實例。

#include <iostream>
#include <cstring>
#include <cstdlib>
#include <fstream>
using namespace std;
 
const int n=26;
typedef struct Trie_node
{
    int count;                    // 統計單詞前綴出現的次數
    struct Trie_node* next[n];   // 指向各個子樹的指針
    bool exist;                  // 標記該結點處是否構成單詞
}TrieNode,*Trie;
 
 
TrieNode *createTrieNode()
{
    TrieNode* node = (TrieNode *)malloc(sizeof(TrieNode));
    node->count = 0;
    node->exist = false;
    memset(node->next , 0 , sizeof(node->next));    // 初始化爲空指針
    return node;
}
 
void Trie_insert(Trie root, char* word)
{
    Trie node = root;
    char *p = word;
    int id;
    while( *p )
    {
        id = *p - 'a';
        if(node->next[id] == NULL)
      {
            node->next[id] = createTrieNode();
         }
        node = node->next[id];  // 每插入一步，至關於有一個新串通過，指針向下移動
        ++p;
        //node->count += 1;      // 這行代碼用於統計每一個單詞前綴出現的次數（也包括統計每一個單詞出現的次數）
    }
    node->exist = true;// 單詞結束的地方標記此處能夠構成一個單詞
    node->count++;
}
 
int Trie_search(Trie root, char* word)
{
    Trie node = root;
    char *p = word;
    int id;
    while( *p )
    {
        id = *p - 'a';
        node = node->next[id];
        ++p;
        if(node == NULL)
    {
        cout<<endl<<word<<"在文件中不存在";
            return 0;
    }
    }
    if(node->exist==true)
        cout<<endl<<word<<"出現了"<<node->count<<"次";
    return node->count;
 
}
 
const int num=5000;
 //產生一個txt文件，模擬字符串
void createStrTXT()
{
    ofstream ofs("word.txt",ios::out);
    for(int i=0;i<num;++i)
    {
    char temp[12]={'\n','\r',rand()%26+97,rand()%26+97,rand()%26+97,rand()%26+97,rand()%26+97,rand()%26+97,rand()%26+97,rand()%26+97,rand()%26+97,'\0'};
 
        char*str=temp;
 
        ofs<<str;
    }
    ofs.close();
}
void establishTrieTree(Trie root)
{
    ifstream ifs("word.txt",ios::in);
    char str[10];
    int i=0;
    while(ifs>>str)
    {
        Trie_insert(root,str);
    cout<<"插入單詞："<<str<<endl;
    i++;
 
    }
    cout<<"總共插入"<<i<<"個單詞";
    ifs.close();
}
 
int main()
{
 
   //初始化root
  Trie root=createTrieNode();
 
  createStrTXT();
 
  establishTrieTree( root);
 
  Trie_search(root,"zxuglsdsm");
 
  return 0;
}

參考：數據結構 練習21-trie的原理分析和應用

---

八、堆

堆類型	做用
最大堆	求前n小
最小堆	求前n大
雙堆	中位數

 

/**
   有一個1G大小的一個文件，裏面每一行是一個詞，詞的大小不超過16字節，內存限制大小是1M。返回頻數最高的100個詞.
   解題思路：使用hash將大文件劃分紅小文件 而後統計小文件中的單詞頻數 而後使用小頂堆 輸出頻數最高的單詞
*/
 
#include<iostream>
#include<string>
#include<cstring>
#include<cstdlib>
#include<cstdio>
using namespace std;
 
#define FILE_NUM 10
#define WORDLEN 30
#define HASHLEN 7303
 
typedef struct node_no_space{
  char *word;
    int number;
    struct node_no_space *next;
}node_no_space, *p_node_no_space;
 
typedef struct node_has_space{
    char word[WORDLEN];
    int number;
    struct node_has_space *next;
}node_has_space, *p_node_has_space;
 
p_node_no_space bin[HASHLEN] = {NULL};
 
void Swap(int *a, int *b) {
    int temp;
    temp = *a;
    *a = *b;
    *b = temp;
}
 
void SwapHeap(node_has_space &node1,node_has_space &node2)
{
    Swap(&node1.number, &node2.number);
    char buffer[WORDLEN];
    strcpy(buffer, node1.word);
    strcpy(node1.word, node2.word);
    strcpy(node2.word, buffer);
}
 
unsigned int Hash(char *p_word) {
    unsigned int index = 0;
    while(*p_word) {
        index += index * 31 + *p_word;
        p_word++;
    }
    return index % HASHLEN;
}
 
void insert_word(char *p_word) {
    unsigned int index = Hash(p_word);
    node_no_space *p;
    for(p = bin[index]; p != NULL; p = p->next) {
            if(strcmp(p_word, p->word) == 0) {
            (p->number)++;
            return;
        }
    }
 
    p = (node_no_space*)malloc(sizeof(node_no_space));
    p->number = 1;
    p->word = (char*)malloc(strlen(p_word) + 1);
    strcpy(p->word, p_word);
    p->next = bin[index];
    bin[index] = p;
}
 
void min_heap(node_has_space *heap, int i, int len) {
    /*int left = 2 * i;
    int right = 2 * i + 1;
    int min_index = 0;
 
    if(left <= len && heap[left].number < heap[i].number) {
        min_index = left;
    } else {
        min_index = i;
    }
 
    if(right <= len && heap[right].number < heap[min_index].number) {
        min_index = right;
    }
    if(min_index != i) {
        Swap(&heap[min_index].number, &heap[i].number);
        char buffer[WORDLEN];
        strcpy(buffer, heap[min_index].word);
        strcpy(heap[min_index].word, heap[i].word);
        strcpy(heap[i].word, buffer);
        min_heap(heap, min_index, len);
    }*/
    int j;
    node_has_space temp = heap[i];
    for(j=2*i;j<=len;j*=2)
    {
        if(j<len&&heap[j+1].number<heap[j].number)
        {
            ++j;
        }
        if(heap[i].number<heap[j].number)
        {
            break;
        }
    SwapHeap(heap[j],heap[i]);//傳引用
    i=j;
    }
    SwapHeap(heap[j],temp);
}
 
void build_min_heap(node_has_space *heap, int len) {
    int index = len / 2;
    int i;
    for(i = index; i >= 1; i--) {
        min_heap(heap, i, len);
    }
}
 
void destroy_bin() {
    node_no_space *p, *q;
    int i = 0;
    while(i < HASHLEN) {
        p = bin[i];
        while(p) {
            q = p->next;
            if(p->word) {
                free(p->word);
                p->word = NULL;
            }
            free(p);
            p = NULL;
            p = q;
        }
        bin[i] = NULL;
        i++;
    }
}
 
void write_to_file(char *path) {
    FILE *out;
    if((out = fopen(path, "w")) == NULL) {
        cout << "error, open " << path << " failed!" << endl;
        return;
    }
    int i;
    node_no_space *p;
    i = 0;
    while(i < HASHLEN) {
        for(p = bin[i]; p != NULL; p = p->next) {
            fprintf(out, "%s %d\n", p->word, p->number);
        }
        i++;
    }
    fclose(out);
    destroy_bin();
}
 
int main() {
    char word[WORDLEN];
    char path[20];
    int number;
    int n = 10;
    unsigned int index = 0;
    int i;
    FILE *fin[10];
    FILE *fout;
    FILE *f_message;
    node_has_space *heap = (node_has_space*)malloc(sizeof(node_has_space) * (n + 1));
    // divide word into n files
    if((f_message = fopen("words.txt", "r")) == NULL) {
        cout << "error, open source file failed!" << endl;
        return 1;
    }
    for(i = 0; i < n; i++) {
        sprintf(path, "tmp%d.txt", i);
        fin[i] = fopen(path, "w");
    }
     while(fscanf(f_message, "%s", word) != EOF) {
        index = Hash(word) % n;
        fprintf(fin[index], "%s\n", word);
    }
    fclose(f_message);
    for(i = 0; i < n; i++) {
        fclose(fin[i]);
    }
 
    // do hash count
    for(i = 0; i < n; i++) {
        sprintf(path, "tmp%d.txt", i);
        fin[i] = fopen(path, "r");
        while(fscanf(fin[i], "%s", word) != EOF) {
            insert_word(word);
        }
        fclose(fin[i]);
        write_to_file(path);
    }
    // heap find
    for(i = 1; i <= n; i++) {
        strcpy(heap[i].word, "");
        heap[i].number = 0;
    }
    build_min_heap(heap, n);
    for(i = 0; i < n; i++) {
        sprintf(path, "tmp%d.txt", i);
        fin[i] = fopen(path, "r");
        while(fscanf(fin[i], "%s %d", word, &number) != EOF) {
            if(number > heap[1].number) {
                heap[1].number = number;
                strcpy(heap[1].word, word);
                min_heap(heap, 1, n);
            }
        }
        fclose(fin[i]);
    }
 
    for(i = 1; i <= n; i++)
        cout << heap[i].word << ":" << heap[i].number << endl;
 
    return 0;
}

實驗數據下載地址

參考：詞頻統計

---

九、雙層桶法

---

十、MapReduce法

參考：用MapReduce大刀砍掉海量數據離線處理問題

 

PS:還沒有徹底整理好，先放出來，慢慢整理。

      海量數據處理

      個人另外一個博客www.flyoung.me搜索引擎收錄很差，因此那邊的文章會同步到博客園上來，如今開始習慣上用markdown寫博客了。