程序员编程艺术：第十章、如何给10^7个数据量的磁盘文件排序

第十章、如何给10^7个数据量的磁盘文件排序

作者:July，yansha，5，编程艺术室。
出处：http://blog.csdn.net/v_JULY_v 。

前奏

经过几天的痛苦沉思，最终决定，把原程序员面试题狂想曲系列正式更名为程序员编程艺术系列，同时，狂想曲创作组更名为编程艺术室。之所以要改名，我们考虑到三点：1、为面试服务不能成为我们最终或最主要的目的，2、我更愿把解答一道道面试题，ACM题等各类程序设计题目的过程，当做一种艺术来看待，3、艺术的提炼本身是一个非常非常艰难的过程，但我们乐意接受这个挑战。

同时，本系列程序编程艺术-算法卷，大致分为三个部分：第一部分--程序设计，大凡如面试题目/ACM题目/poj的题目等各类程序设计的题，只要是好的，值得设计或深究的题目，我们都不拒绝。同时，紧扣实际，不断寻找更高效的算法解决实际问题。第二部分--算法研究，主要以我个人此前写的原创作品-十三个经典算法研究系列为题材，力争通俗易懂，详略得当的剖析各类经典的算法，并予以编程实现。第三部分--编码素养，主要包括程序员编码过程中一些编码规范等各类及其需要注意的问题。

如果有可能的话，此TAOPP系列将采取TAOCP那样的形式，出第一卷、第二卷、...。编程艺术来自哪里?编程采取合适的数据结构?寻求更高效的算法?或者，好的编码规范?希望，本TAOPP系列最终能给你一个完整的答复。

ok，如果任何人对本编程艺术系列有任何意见，或发现了本编程艺术系列任何问题，漏洞，bug，欢迎随时提出，我们将虚心接受并感激不尽，以为他人创造更好的价值，更好的服务。

第一节、如何给磁盘文件排序
问题描述：
输入：一个最多含有n个不重复的正整数（也就是说可能含有少于n个不重复正整数）的文件，其中每个数都小于等于n，且n=10^7。
输出：得到按从小到大升序排列的包含所有输入的整数的列表。
条件：最多有大约1MB的内存空间可用，但磁盘空间足够。且要求运行时间在5分钟以下，10秒为最佳结果。

分析：下面咱们来一步一步的解决这个问题，
1、归并排序。你可能会想到把磁盘文件进行归并排序，但题目要求你只有1MB的内存空间可用，所以，归并排序这个方法不行。
2、位图方案。熟悉位图的朋友可能会想到用位图来表示这个文件集合。例如正如编程珠玑一书上所述，用一个20位长的字符串来表示一个所有元素都小于20的简单的非负整数集合，边框用如下字符串来表示集合{1,2,3,5,8,13}：

0 1 1 1 0 1 0 0 1 0 0 0 0 1 0 0 0 0 0 0

上述集合中各数对应的位置则置1，没有对应的数的位置则置0。

参考编程珠玑一书上的位图方案，针对我们的10^7个数据量的磁盘文件排序问题，我们可以这么考虑，由于每个7位十进制整数表示一个小于1000万的整数。我们可以使用一个具有1000万个位的字符串来表示这个文件，其中，当且仅当整数i在文件中存在时，第i位为1。采取这个位图的方案是因为我们面对的这个问题的特殊性：1、输入数据限制在相对较小的范围内，2、数据没有重复，3、其中的每条记录都是单一的整数，没有任何其它与之关联的数据。
所以，此问题用位图的方案分为以下三步进行解决：

第一步，将所有的位都置为0，从而将集合初始化为空。
第二步，通过读入文件中的每个整数来建立集合，将每个对应的位都置为1。
第三步，检验每一位，如果该位为1，就输出对应的整数。
经过以上三步后，产生有序的输出文件。令n为位图向量中的位数（本例中为1000 0000），程序可以用伪代码表示如下：

//copyright@ yansha//July、2010.05.30。//位图方案解决10^7个数据量的文件的排序问题//如果有重复的数据，那么只能显示其中一个 其他的将被忽略#include <iostream>#include <bitset>#include <assert.h>#include <time.h>using namespace std;const int max_each_scan = 5000000;int main(){clock_t begin = clock();bitset<max_each_scan> bit_map;bit_map.reset();// open the file with the unsorted dataFILE *fp_unsort_file = fopen("data.txt", "r");assert(fp_unsort_file);int num;// the first time scan to sort the data between 0 - 4999999while (fscanf(fp_unsort_file, "%d ", &num) != EOF){if (num < max_each_scan)bit_map.set(num, 1);}FILE *fp_sort_file = fopen("sort.txt", "w");assert(fp_sort_file);int i;// write the sorted data into filefor (i = 0; i < max_each_scan; i++){if (bit_map[i] == 1)fprintf(fp_sort_file, "%d ", i);}// the second time scan to sort the data between 5000000 - 9999999int result = fseek(fp_unsort_file, 0, SEEK_SET);if (result)cout << "fseek failed!" << endl;else{bit_map.reset();while (fscanf(fp_unsort_file, "%d ", &num) != EOF){if (num >= max_each_scan && num < 10000000){num -= max_each_scan;bit_map.set(num, 1);}}for (i = 0; i < max_each_scan; i++){if (bit_map[i] == 1)fprintf(fp_sort_file, "%d ", i + max_each_scan);}}clock_t end = clock();cout<<"用位图的方法，耗时："<<endl;cout << (end - begin) / CLK_TCK << "s" << endl;fclose(fp_sort_file);fclose(fp_unsort_file);return 0;}

而后测试了一下上述程序的运行时间，采取位图方案耗时14s，即14000ms：



本章中，生成大数据量（1000w）的程序如下，下文第二节的多路归并算法的c++实现和第三节的磁盘文件排序的编程实现中，生成的1000w数据量也是用本程序产生的，且本章内生成的1000w数据量的数据文件统一命名为“data.txt”。

//copyright@ 纯净的天空 && yansha//5、July，updated，2010.05.28。#include <iostream>#include <ctime>#include <fstream>//#include "ExternSort.h"using namespace std;//使用多路归并进行外排序的类//ExternSort.h/** 大数据量的排序* 多路归并排序* 以千万级整数从小到大排序为例* 一个比较简单的例子，没有建立内存缓冲区*/#ifndef EXTERN_SORT_H#define EXTERN_SORT_H#include <cassert>class ExternSort{public:void sort(){time_t start = time(NULL);//将文件内容分块在内存中排序，并分别写入临时文件int file_count = memory_sort();//归并临时文件内容到输出文件merge_sort(file_count);time_t end = time(NULL);printf("total time:%f/n", (end - start) * 1000.0/ CLOCKS_PER_SEC);}//input_file:输入文件名//out_file:输出文件名//count: 每次在内存中排序的整数个数ExternSort(const char *input_file, const char * out_file, int count){m_count = count;m_in_file = new char[strlen(input_file) + 1];strcpy(m_in_file, input_file);m_out_file = new char[strlen(out_file) + 1];strcpy(m_out_file, out_file);}virtual ~ExternSort(){delete [] m_in_file;delete [] m_out_file;}private:int m_count; //数组长度char *m_in_file;   //输入文件的路径char *m_out_file; //输出文件的路径protected:int read_data(FILE* f, int a[], int n){int i = 0;while(i < n && (fscanf(f, "%d", &a[i]) != EOF)) i++;printf("read:%d integer/n", i);return i;}void write_data(FILE* f, int a[], int n){for(int i = 0; i < n; ++i)fprintf(f, "%d ", a[i]);}char* temp_filename(int index){char *tempfile = new char[100];sprintf(tempfile, "temp%d.txt", index);return tempfile;}static int cmp_int(const void *a, const void *b){return *(int*)a - *(int*)b;}int memory_sort(){FILE* fin = fopen(m_in_file, "rt");int n = 0, file_count = 0;int *array = new int[m_count];//每读入m_count个整数就在内存中做一次排序，并写入临时文件while(( n = read_data(fin, array, m_count)) > 0){qsort(array, n, sizeof(int), cmp_int);   //这里，调用了库函数阿，在第四节的c实现里，不再调用qsort。char *fileName = temp_filename(file_count++);FILE *tempFile = fopen(fileName, "w");free(fileName);write_data(tempFile, array, n);fclose(tempFile);}delete [] array;fclose(fin);return file_count;}void merge_sort(int file_count){if(file_count <= 0) return;//归并临时文件FILE *fout = fopen(m_out_file, "wt");FILE* *farray = new FILE*[file_count];int i;for(i = 0; i < file_count; ++i){char* fileName = temp_filename(i);farray[i] = fopen(fileName, "rt");free(fileName);}int *data = new int[file_count];//存储每个文件当前的一个数字bool *hasNext = new bool[file_count];//标记文件是否读完memset(data, 0, sizeof(int) * file_count);memset(hasNext, 1, sizeof(bool) * file_count);for(i = 0; i < file_count; ++i){if(fscanf(farray[i], "%d", &data[i]) == EOF)//读每个文件的第一个数到data数组hasNext[i] = false;}while(true){//求data中可用的最小的数字，并记录对应文件的索引int min = data[0];int j = 0;while (j < file_count && !hasNext[j])j++;if (j >= file_count)  //没有可取的数字，终止归并break;for(i = j + 1; i < file_count; ++i){if(hasNext[i] && min > data[i]){min = data[i];j = i;}}if(fscanf(farray[j], "%d", &data[j]) == EOF) //读取文件的下一个元素hasNext[j] = false;fprintf(fout, "%d ", min);}delete [] hasNext;delete [] data;for(i = 0; i < file_count; ++i){fclose(farray[i]);}delete [] farray;fclose(fout);}};#endif//测试主函数文件/** 大文件排序* 数据不能一次性全部装入内存* 排序文件里有多个整数，整数之间用空格隔开*/const unsigned int count = 10000000; // 文件里数据的行数const unsigned int number_to_sort = 1000000; //在内存中一次排序的数量const char *unsort_file = "unsort_data.txt"; //原始未排序的文件名const char *sort_file = "sort_data.txt"; //已排序的文件名void init_data(unsigned int num); //随机生成数据文件int main(int argc, char* *argv){srand(time(NULL));init_data(count);ExternSort extSort(unsort_file, sort_file, number_to_sort);extSort.sort();system("pause");return 0;}void init_data(unsigned int num){FILE* f = fopen(unsort_file, "wt");for(int i = 0; i < num; ++i)fprintf(f, "%d ", rand());fclose(f);}

程序测试：读者可以继续用小文件小数据量进一步测试。


第三节、磁盘文件排序的编程实现

ok，接下来，我们来编程实现上述磁盘文件排序的问题，本程序由两部分构成：
1、内存排序
由于要求的可用内存为1MB，那么每次可以在内存中对250K的数据进行排序，然后将有序的数写入硬盘。
那么10M的数据需要循环40次，最终产生40个有序的文件。
2、归并排序

将每个文件最开始的数读入(由于有序，所以为该文件最小数)，存放在一个大小为40的first_data数组中；
选择first_data数组中最小的数min_data，及其对应的文件索引index；
将first_data数组中最小的数写入文件result，然后更新数组first_data(根据index读取该文件下一个数代替min_data)；
判断是否所有数据都读取完毕，否则返回2。
所以，本程序按顺序分两步，第一步、Memory Sort，第二步、Merge Sort。程序的流程图，如下图所示（感谢F的绘制）。





然后，编写的完整代码如下：

//copyright@ 555//July、2011.05.29。#include <assert.h>#include <time.h> #include <stdio.h>   #include <memory.h>#include <stdlib.h>void swap_int(int* a,int* b){    int c;    c = *a;    *a = *b;    *b = c;}//插入排序void InsertionSort(int A[],int N){    int j,p;    int tmp;    for(p = 1; p < N; p++)    {        tmp = A[p];for(j = p;j > 0 && A[j - 1] >tmp;j--)        {            A[j] = A[j - 1];        }        A[j] = tmp;    }}//三数取中分割法int Median3(int A[],int Left,int Right){int Center = (Left + Right) / 2;if (A[Left] > A[Center])swap_int(&A[Left],&A[Center]);if (A[Left] > A[Right])swap_int(&A[Left],&A[Right]);if (A[Center] > A[Right])swap_int(&A[Center],&A[Right]);swap_int(&A[Center],&A[Right - 1]);return A[Right - 1];}//快速排序void QuickSort(int A[],int Left,int Right){int i,j;int Pivot;const int Cutoff = 3;if (Left + Cutoff <= Right){Pivot = Median3(A,Left,Right);i = Left;j = Right - 1;while (1){while(A[++i] < Pivot){;}while(A[--j] > Pivot){;}if (i < j)swap_int(&A[i],&A[j]);elsebreak;}swap_int(&A[i],&A[Right - 1]); QuickSort(A,Left,i - 1);QuickSort(A,i + 1,Right);}else{InsertionSort(A+Left,Right - Left + 1);}}//const int  KNUM  = 40;        //分块数//const int  NUMBER = 10000000; //输入文件最大读取的整数的个数//为了便于测试，我决定改成小文件小数据量进行测试。const int  KNUM  = 4;        //分块数const int  NUMBER = 100; //输入文件最大读取的整数的个数const char *in_file = "infile.txt"; const char *out_file = "outfile.txt";//#define OUTPUT_OUT_FILE_DATA//数据量大的时候,没必要把所有的数全部打印出来，所以可以把上面这句注释掉。void  gen_infile(int n){int i;FILE *f = fopen(in_file, "wt"); for(i = 0;i < n; i++)fprintf(f,"%d ",rand());fclose(f);}int  read_data(FILE *f,int a[],int n){int i = 0;while ((i < n) && (fscanf(f,"%d",&a[i]) != EOF))  i++;printf("read: %d integer/n",i);return i;}void  write_data(FILE *f,int a[],int n){int i;for(i = 0; i< n;i++)fprintf(f,"%d ",a[i]);}char* temp_filename(int index){char *tempfile = (char*) malloc(64*sizeof(char));assert(tempfile);sprintf(tempfile, "temp%d.txt", index);return tempfile;}//K路串行读取void k_num_read(void){char* filename;int i,cnt,*array;FILE* fin;FILE* tmpfile;//计算knum,每路应读取的整数个数int n = NUMBER/KNUM;if (n * KNUM < NUMBER)n++;//建立存储分块读取的数据的数组array = (int*)malloc(n * sizeof(int));assert(array);//打开输入文件fin = fopen(in_file,"rt");i = 0;//分块循环读取数据,并写入硬盘上的临时文件while ( (cnt = read_data(fin,array,n))>0){//对每次读取的数据,先进行快速排序,然后写入硬盘上的临时文件QuickSort(array,0,cnt - 1);filename = temp_filename(i++);tmpfile = fopen(filename,"w");free(filename);write_data(tmpfile,array,cnt);fclose(tmpfile);}assert(i == KNUM);;//没有生成K路文件时进行诊断//关闭输入文件句柄和临时存储数组fclose(fin);free(array);}//k路合并(败者树)void k_num_merge(void){FILE *fout;FILE **farray;char *filename;int  *data;char *hasNext;int i,j,m,min;#ifdef OUTPUT_OUT_FILE_DATAint id;#endif//打开输出文件fout = fopen(out_file,"wt");//打开各路临时分块文件farray = (FILE**)malloc(KNUM*sizeof(FILE*));assert(farray);for(i = 0; i< KNUM;i++){filename = temp_filename(i);farray[i] = fopen(filename,"rt");free(filename);}//建立KNUM个元素的data,hasNext数组,存储K路文件的临时数组和读取结束状态data = (int*)malloc(KNUM*sizeof(int));assert(data);hasNext = (char*)malloc(sizeof(char)*KNUM);assert(hasNext);memset(data, 0, sizeof(int) * KNUM);memset(hasNext, 1, sizeof(char) * KNUM);//读K路文件先读取第一组数据,并对读取结束的各路文件设置不可再读状态for(i = 0; i < KNUM; i++){if(fscanf(farray[i], "%d", &data[i]) == EOF){hasNext[i] = 0;}}//读取各路文件,利用败者树从小到大输出到输出文件#ifdef OUTPUT_OUT_FILE_DATAid = 0;#endifj  = 0;F_LOOP:if (j < KNUM)    //以下这段代码嵌套过深，日后应尽量避免此类问题。{while(1==1){min = data[j];m = j;for(i = j+1; i < KNUM; i++){if(hasNext[i] == 1  && min > data[i]){min = data[i];m = i;}}if(fscanf(farray[m], "%d", &data[m]) == EOF) {hasNext[m] = 0;}fprintf(fout, "%d ", min);#ifdef OUTPUT_OUT_FILE_DATAprintf("fout :%d  %d/n",++id,min);#endifif (m == j && hasNext[m] == 0){for (i = j+1; i < KNUM; i++){if (hasNext[m] != hasNext[i]){m = i;//第i个文件未读完,从第i个继续往下读break;}}if (m != j){j = m;goto F_LOOP;}break;}}}//关闭分配的数据和数组    free(hasNext);    free(data);        for(i = 0; i < KNUM; ++i)    {        fclose(farray[i]);    }    free(farray);    fclose(fout);}int main()    {   time_t start = time(NULL),end,start_read,end_read,start_merge,end_merge;gen_infile(NUMBER);    end = time(NULL);    printf("gen infile data time:%f/n", (end - start) * 1000.0/ CLOCKS_PER_SEC);start_read = time(NULL);k_num_read();    end_read = time(NULL);    printf("k_num_read time:%f/n", (end_read - start_read) * 1000.0/ CLOCKS_PER_SEC);start_merge = time(NULL);k_num_merge();    end_merge = time(NULL);    printf("k_num_merge time:%f/n", (end_merge - start_merge) * 1000.0/ CLOCKS_PER_SEC);    end = time(NULL);    printf("total time:%f/n", (end - start) * 1000.0/ CLOCKS_PER_SEC);    return 0;   }

程序测试：

在此，我们先测试下对10000000个数据的文件进行40趟排序，然后再对100个数据的文件进行4趟排序（读者可进一步测试）。如弄几组小点的数据,输出ID和数据到屏幕，再看程序运行效果。

10个数, 4组
40个数, 5组
55个数, 6组
100个数, 7组





（备注：1、以上所有各节的程序运行环境为windows xp + vc6.0 + e5200 cpu 2.5g主频，2、感谢5为本文程序所作的大量测试工作）

全文总结：

1、关于本章中位图和多路归并两种方案的时间复杂度及空间复杂度的比较，如下：

时间复杂度 空间复杂度
位图 O(N) 0.625M
多位归并 O(Nlogn) 1M

（多路归并，时间复杂度为O（k*n/k*logn/k ），严格来说，还要加上读写磁盘的时间，而此算法绝大部分时间也是浪费在这上面）

2、bit-map

适用范围：可进行数据的快速查找，判重，删除，一般来说数据范围是int的10倍以下
基本原理及要点：使用bit数组来表示某些元素是否存在，比如8位电话号码
扩展：bloom filter可以看做是对bit-map的扩展

问题实例：
1)已知某个文件内包含一些电话号码，每个号码为8位数字，统计不同号码的个数。
8位最多99 999 999，大概需要99m个bit，大概10几m字节的内存即可。
2)2.5亿个整数中找出不重复的整数的个数，内存空间不足以容纳这2.5亿个整数。

将bit-map扩展一下，用2bit表示一个数即可，0表示未出现，1表示出现一次，2表示出现2次及以上。或者我们不用2bit来进行表示，我们用两个bit-map即可模拟实现这个2bit-map。

3、[外排序适用范围]大数据的排序，去重基本原理及要点：外排序的归并方法，置换选择败者树原理，最优归并树扩展。问题实例：1).有一个1G大小的一个文件，里面每一行是一个词，词的大小不超过16个字节，内存限制大小是1M。返回频数最高的100个词。这个数据具有很明显的特点，词的大小为16个字节，但是内存只有1m做hash有些不够，所以可以用来排序。内存可以当输入缓冲区使用。

4、海量数据处理

有关海量数据处理的方法或面试题可参考此文，十道海量数据处理面试题与十个方法大总结。日后，会逐步实现这十个处理海量数据的方法。同时，送给各位一句话，解决问题的关键在于熟悉一个算法，而不是某一个问题。熟悉了一个算法，便通了一片题目。

本章完。

版权所有，本人对本blog内所有任何内容享有版权及著作权。网络转载，请以链接形式注明出处。

作者：v_JULY_v 发表于2011-5-28 16:25:00 原文链接

阅读：6413 评论：93 查看评论