海量数据处理

基本方法

1、Hash法

参考：

2、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实现起来更加简单。

3、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;
       }
    }  
}

参考：

4、数据库优化法

5、倒排索引法

6、外排序法

7、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的原理分析和应用

8、堆

堆类型	作用
最大堆	求前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;
}