定义问题

      外部排序指的是大文件的排序，即待排序的记录存储在外存储器上，待排序的文件无法一次装入内存，需要在内存和外部存储器之间进行多次数据交换，以达到排序整个文件的目的。外部排序最常用的算法是多路归并排序，即将原文件分解成多个能够一次性装入内存的部分，分别把每一部分调入内存完成排序。然后，对已经排序的子文件进行多路归并排序。

处理过程

  （1）按可用内存的大小，把外存上含有n个记录的文件分成若干个长度为L的子文件，把这些子文件依次读入内存，并利用有效的内部排序方法对它们进行排序，再将排序后得到的有序子文件重新写入外存；

  （2）对这些有序子文件逐趟归并，使其逐渐由小到大，直至得到整个有序文件为止。

   先从一个例子来看外排序中的归并是如何进行的？
  假设有一个含10000 个记录的文件，首先通过10 次内部排序得到10 个初始归并段R1～R10 ，其中每一段都含1000 个记录。然后对它们作如图10.11 所示的两两归并，直至得到一个有序文件为止 如下图

  

 

 

多路归并排序算法以及败者树

    多路归并排序算法在常见数据结构书中都有涉及。从2路到多路（k路），增大k可以减少外存信息读写时间，但k个归并段中选取最小的记录需要比较k-1次，为得到u个记录的一个有序段共需要(u-1)(k-1)次，若归并趟数为s次，那么对n个记录的文件进行外排时，内部归并过程中进行的总的比较次数为s(n-1)(k-1)，也即(向上取整)(logkm)(k-1)(n-1)=(向上取整)(log2m/log2k)(k-1)(n-1)，而(k-1)/log2k随k增而增因此内部归并时间随k增长而增长了，抵消了外存读写减少的时间，这样做不行，由此引出了“败者树”tree of loser的使用。在内部归并过程中利用败者树将k个归并段中选取最小记录比较的次数降为(向上取整)(log2k)次使总比较次数为(向上取整)(log2m)(n-1)，与k无关。

    败者树是完全二叉树，因此数据结构可以采用一维数组。其元素个数为k个叶子结点、k-1个比较结点、1个冠军结点共2k个。ls[0]为冠军结点，ls[1]--ls[k-1]为比较结点，ls[k]--ls[2k-1]为叶子结点（同时用另外一个指针索引b[0]--b[k-1]指向）。另外bk为一个附加的辅助空间，不属于败者树，初始化时存着MINKEY的值。

    多路归并排序算法的过程大致为：

 

   1）：首先将k个归并段中的首元素关键字依次存入b[0]--b[k-1]的叶子结点空间里，然后调用CreateLoserTree创建败者树，创建完毕之后最小的关键字下标（即所在归并段的序号）便被存入ls[0]中。然后不断循环：

   2）把ls[0]所存最小关键字来自于哪个归并段的序号得到为q，将该归并段的首元素输出到有序归并段里，然后把下一个元素关键字放入上一个元素本来所在的叶子结点b[q]中，调用Adjust顺着b[q]这个叶子结点往上调整败者树直到新的最小的关键字被选出来，其下标同样存在ls[0]中。循环这个操作过程直至所有元素被写到有序归并段里。

   四、伪代码：

void Adjust(LoserTree &ls, int s)
/*从叶子结点b[s]到根结点的父结点ls[0]调整败者树*/
{  int t, temp;
   t=(s+K)/2;          /*t为b[s]的父结点在败者树中的下标，K是归并段的个数*/
   while(t>0)                         /*若没有到达树根，则继续*/
   {     if(b[s]>b[ls[t]])        /*与父结点指示的数据进行比较*/
               {  /*ls[t]记录败者所在的段号，s指示新的胜者，胜者将去参加更上一层的比较*/
                  temp=s;
                  s=ls[t];
                  ls[t]=temp; 
                }
           t=t/2;                     /*向树根退一层，找到父结点*/
   }
  ls[0]=s;                           /*ls[0]记录本趟最小关键字所在的段号*/
}

void K_merge( int ls[K])
/*ls[0]~ls[k-1]是败者树的内部比较结点。b[0]~b[k-1]分别存储k个初始归并段的当前记录*/
/*函数Get_next(i)用于从第i个归并段读取并返回当前记录*/
{   int b[K+1),i,q;
     for(i=0; i<K;i++)                
     {   b[i]=Get_next(i);           /*分别读取K个归并段的第一个关键字*/  } 
     b[K]=MINKEY;                        /*创建败者树*/
     for(i=0; i<K ; i++)                    /*设置ls中的败者初值*/
           ls[i]=K;
     for(i=K-1 ; i>=0 ; i--)                /*依次从b[K-1]……b[0]出发调整败者*/
          Adjust(ls , i);             /*败者树创建完毕，最小关键字序号存入ls[0]
     while(b[ls[0]] !=MAXKEY )
     {   q=ls[0];                        /*q为当前最小关键字所在的归并段*/
          prinftf("%d",b[q]);
          b[q]=Get_next(q);
          Adjust(ls,q);                /*q为调整败者树后，选择新的最小关键字*/
     }
} 

如下图，一个详细的过程。2个子结点比较后的败者放入它们的父结点，而胜者送到它们父结点的父节点去再作比较，这才是败者树。b[0]放的是最终的胜者。

 

 

胜者树与败者树  

 

 

       胜者树和败者树都是完全二叉树，是树形选择排序的一种变型。每个叶子结点相当于一个选手，每个中间结点相当于一场比赛，每一层相当于一轮比赛。

 

      不同的是，胜者树的中间结点记录的是胜者的标号；而败者树的中间结点记录的败者的标号。

 

       胜者树与败者树可以在log(n)的时间内找到最值。任何一个叶子结点的值改变后，利用中间结点的信息，还是能够快速地找到最值。在k路归并排序中经常用到。

 

胜者树

      

       胜者树的一个优点是，如果一个选手的值改变了，可以很容易地修改这棵胜者树。只需要沿着从该结点到根结点的路径修改这棵二叉树，而不必改变其他比赛的结果。

 

Fig. 1

Fig.1是一个胜者树的示例。规定数值小者胜。

1.         b3 PK b4，b3胜b4负，内部结点ls[4]的值为3；

2.         b3 PK b0，b3胜b0负，内部结点ls[2]的值为3；

3.         b1 PK b2，b1胜b2负，内部结点ls[3]的值为1；

4.         b3 PK b1，b3胜b1负，内部结点ls[1]的值为3。.

当Fig. 1中叶子结点b3的值变为11时，重构的胜者树如Fig. 2所示。

1.         b3 PK b4，b3胜b4负，内部结点ls[4]的值为3；

2.         b3 PK b0，b0胜b3负，内部结点ls[2]的值为0；

3.         b1 PK b2，b1胜b2负，内部结点ls[3]的值为1；

4.         b0 PK b1，b1胜b0负，内部结点ls[1]的值为1。.

Fig. 2

 

 

败者树

 

       败者树是胜者树的一种变体。在败者树中，用父结点记录其左右子结点进行比赛的败者，而让胜者参加下一轮的比赛。败者树的根结点记录的是败者，需要加一个结点来记录整个比赛的胜利者。采用败者树可以简化重构的过程。

 

Fig. 3

Fig. 3是一棵败者树。规定数大者败。

1.         b3 PK b4，b3胜b4负，内部结点ls[4]的值为4；

2.         b3 PK b0，b3胜b0负，内部结点ls[2]的值为0；

3.         b1 PK b2，b1胜b2负，内部结点ls[3]的值为2；

4.         b3 PK b1，b3胜b1负，内部结点ls[1]的值为1；

5.         在根结点ls[1]上又加了一个结点ls[0]=3，记录的最后的胜者。

败者树重构过程如下：

·            将新进入选择树的结点与其父结点进行比赛：将败者存放在父结点中；而胜者再与上一级的父结点比较。

·            比赛沿着到根结点的路径不断进行，直到ls[1]处。把败者存放在结点ls[1]中，胜者存放在ls[0]中。

Fig. 4

       Fig. 4是当b3变为13时，败者树的重构图。

 

       注意，败者树的重构跟胜者树是不一样的，败者树的重构只需要与其父结点比较。对照Fig. 3来看，b3与结点ls[4]的原值比较，ls[4]中存放的原值是结点4，即b3与b4比较，b3负b4胜，则修改ls[4]的值为结点3。同理，以此类推，沿着根结点不断比赛，直至结束。

 

        由上可知，败者树简化了重构。败者树的重构只是与该结点的父结点的记录有关，而胜者树的重构还与该结点的兄弟结点有关。

 

 

 

败者树 多路平衡归并外部排序

 

 

 

 

外部排序的基本思路

假设有一个72KB的文件，其中存储了18K个整数，磁盘中物理块的大小为4KB，将文件分成18组，每组刚好4KB。

首先通过18次内部排序，把18组数据排好序，得到初始的18个归并段R1~R18，每个归并段有1024个整数。

然后对这18个归并段使用4路平衡归并排序：

第1次归并：产生5个归并段

R11   R12    R13    R14    R15

其中

R11是由{R1,R2,R3,R4}中的数据合并而来

R12是由{R5,R6,R7,R8}中的数据合并而来

R13是由{R9,R10,R11,R12}中的数据合并而来

R14是由{R13,R14,R15,R16}中的数据合并而来

R15是由{R17,R18}中的数据合并而来

把这5个归并段的数据写入5个文件：

foo_1.dat    foo_2.dat    foo_3.dat     foo_4.dat     foo_5.dat

 

第2次归并：从第1次归并产生的5个文件中读取数据，合并，产生2个归并段

R21  R22

其中R21是由{R11,R12,R13,R14}中的数据合并而来

其中R22是由{R15}中的数据合并而来

把这2个归并段写入2个文件

bar_1.dat   bar_2.dat

 

第3次归并：从第2次归并产生的2个文件中读取数据，合并，产生1个归并段

R31

R31是由{R21,R22}中的数据合并而来

把这个文件写入1个文件

foo_1.dat

此即为最终排序好的文件。

 

使用败者树加快合并排序

外部排序最耗时间的操作时磁盘读写，对于有m个初始归并段，k路平衡的归并排序，磁盘读写次数为

|logkm|，可见增大k的值可以减少磁盘读写的次数，但增大k的值也会带来负面效应，即进行k路合并

的时候会增加算法复杂度，来看一个例子。

把n个整数分成k组，每组整数都已排序好，现在要把k组数据合并成1组排好序的整数，求算法复杂度

u1: xxxxxxxx

u2: xxxxxxxx

u3: xxxxxxxx

.......

uk: xxxxxxxx

算法的步骤是：每次从k个组中的首元素中选一个最小的数，加入到新组，这样每次都要比较k-1次，故

算法复杂度为O((n-1)*(k-1))，而如果使用败者树，可以在O(logk)的复杂度下得到最小的数，算法复杂

度将为O((n-1)*logk)， 对于外部排序这种数据量超大的排序来说，这是一个不小的提高。

 

关于败者树的创建和调整，可以参考清华大学《数据结构-C语言版》

 

产生二进制测试数据

打开Linux终端，输入命令

dd if=/dev/urandom of=random.dat bs=1M count=512

 这样在当前目录下产生一个512M大的二进制文件，文件内的数据是随机的，读取文件，每4个字节

看成1个整数，相当于得到128M个随机整数。

 

程序实现

 

[cpp] view plain copy

1. #include <assert.h>  
2. #include <fcntl.h>  
3. #include <stdio.h>  
4. #include <stdlib.h>  
5. #include <string.h>  
6. #include <unistd.h>  
7.   
8. #include <sys/time.h>  
9. #include <sys/types.h>  
10. #include <sys/stat.h>  
11.   
12. #define MAX_INT ~(1<<31)  
13. #define MIN_INT 1<<31  
14.   
15. //#define DEBUG  
16.   
17. #ifdef DEBUG  
18. #define debug(...) debug( __VA_ARGS__)   
19. #else  
20. #define debug(...)  
21. #endif  
22.   
23. #define MAX_WAYS 100  
24.   
25. typedef struct run_t {  
26.     int *buf;       /* 输入缓冲区 */  
27.     int length;     /* 缓冲区当前有多少个数 */  
28.     int offset;     /* 缓冲区读到了文件的哪个位置 */  
29.     int idx;        /* 缓冲区的指针 */  
30. } run_t;  
31.   
32. static unsigned int K;              /* K路合并 */  
33. static unsigned int BUF_PAGES;      /* 缓冲区有多少个page */  
34. static unsigned int PAGE_SIZE;      /* page的大小 */  
35. static unsigned int BUF_SIZE;       /* 缓冲区的大小, BUF_SIZE = BUF_PAGES*PAGE_SIZE */  
36.   
37. static int *buffer;                 /* 输出缓冲区 */  
38.   
39. static char input_prefix[] = "foo_";  
40. static char output_prefix[] = "bar_";  
41.   
42. static int ls[MAX_WAYS];            /* loser tree */  
43.   
44. void swap(int *p, int *q);  
45. int partition(int *a, int s, int t);  
46. void quick_sort(int *a, int s, int t);  
47. void adjust(run_t ** runs, int n, int s);  
48. void create_loser_tree(run_t **runs, int n);  
49. long get_time_usecs();  
50. void k_merge(run_t** runs, char* input_prefix, int num_runs, int base, int n_merge);  
51. void usage();  
52.   
53.   
54. int main(int argc, char **argv)  
55. {  
56.     char                filename[100];  
57.     unsigned int    data_size;  
58.     unsigned int    num_runs;               /* 这轮迭代时有多少个归并段 */  
59.     unsigned int    num_merges;             /* 这轮迭代后产生多少个归并段 num_merges = num_runs/K */  
60.     unsigned int    run_length;             /* 归并段的长度，指数级增长 */  
61.     unsigned int    num_runs_in_merge;      /* 一般每个merge由K个runs合并而来，但最后一个merge可能少于K个runs */  
62.     int                 fd, rv, i, j, bytes;  
63.     struct stat         sbuf;  
64.   
65.     if (argc != 3) {  
66.         usage();  
67.         return 0;  
68.     }  
69.     long start_usecs = get_time_usecs();  
70.   
71.     strcpy(filename, argv[1]);  
72.     fd = open(filename, O_RDONLY);  
73.     if (fd < 0) {  
74.         printf("can't open file %s\n", filename);  
75.         exit(0);  
76.     }  
77.     rv = fstat(fd, &sbuf);  
78.     data_size = sbuf.st_size;  
79.   
80.     K = atoi(argv[2]);  
81.     PAGE_SIZE = 4096;                           /* page = 4KB */  
82.     BUF_PAGES = 32;  
83.     BUF_SIZE = PAGE_SIZE*BUF_PAGES;  
84.     num_runs = data_size / PAGE_SIZE;           /* 初始时的归并段数量，每个归并段有4096 byte, 即1024个整数 */  
85.     buffer = (int *)malloc(BUF_SIZE);  
86.   
87.     run_length = 1;  
88.     run_t **runs = (run_t **)malloc(sizeof(run_t *)*(K+1));  
89.     for (i = 0; i < K; i++) {  
90.         runs[i] = (run_t *)malloc(sizeof(run_t));  
91.         runs[i]->buf = (int *)calloc(1, BUF_SIZE+4);  
92.     }  
93.     while (num_runs > 1) {  
94.         num_merges = num_runs / K;  
95.         int left_runs = num_runs % K;  
96.         if(left_runs > 0) num_merges++;  
97.         for (i = 0; i < num_merges; i++) {  
98.             num_runs_in_merge = K;  
99.             if ((i+1) == num_merges && left_runs > 0) {  
100.                 num_runs_in_merge = left_runs;  
101.             }  
102.             int base = 0;  
103.             printf("Merge %d of %d,%d ways\n", i, num_merges, num_runs_in_merge);  
104.             for (j = 0; j < num_runs_in_merge; j++) {  
105.                 if (run_length == 1) {  
106.                     base = 1;  
107.                     bytes = read(fd, runs[j]->buf, PAGE_SIZE);  
108.                     runs[j]->length = bytes/sizeof(int);  
109.                     quick_sort(runs[j]->buf, 0, runs[j]->length-1);  
110.                 } else {  
111.                     snprintf(filename, 20, "%s%d.dat", input_prefix, i*K+j);  
112.                     int infd = open(filename, O_RDONLY);  
113.                     bytes = read(infd, runs[j]->buf, BUF_SIZE);  
114.                     runs[j]->length = bytes/sizeof(int);  
115.                     close(infd);      
116.                 }  
117.                 runs[j]->idx = 0;  
118.                 runs[j]->offset = bytes;  
119.             }  
120.             k_merge(runs, input_prefix, num_runs_in_merge, base, i);  
121.         }  
122.   
123.         strcpy(filename, output_prefix);  
124.         strcpy(output_prefix, input_prefix);  
125.         strcpy(input_prefix, filename);  
126.   
127.         run_length *= K;  
128.         num_runs = num_merges;  
129.     }  
130.   
131.     for (i = 0; i < K; i++) {  
132.         free(runs[i]->buf);  
133.         free(runs[i]);  
134.     }  
135.     free(runs);  
136.     free(buffer);  
137.     close(fd);  
138.   
139.     long end_usecs = get_time_usecs();  
140.     double secs = (double)(end_usecs - start_usecs) / (double)1000000;  
141.     printf("Sorting took %.02f seconds.\n", secs);  
142.     printf("sorting result saved in %s%d.dat.\n", input_prefix, 0);  
143.   
144.     return 0;  
145. }  
146.   
147. void k_merge(run_t** runs, char* input_prefix, int num_runs, int base, int n_merge)  
148. {  
149.     int bp, bytes, output_fd;  
150.     int live_runs = num_runs;  
151.     run_t *mr;  
152.     char filename[20];  
153.   
154.     bp = 0;  
155.     create_loser_tree(runs, num_runs);  
156.   
157.     snprintf(filename, 100, "%s%d.dat", output_prefix, n_merge);  
158.     output_fd = open(filename, O_CREAT|O_WRONLY|O_TRUNC,   
159.             S_IRWXU|S_IRWXG);  
160.     if (output_fd < 0) {  
161.         printf("create file %s fail\n", filename);  
162.         exit(0);  
163.     }  
164.   
165.     while (live_runs > 0) {  
166.         mr = runs[ls[0]];  
167.         buffer[bp++] = mr->buf[mr->idx++];  
168.         // 输出缓冲区已满  
169.         if (bp*4 == BUF_SIZE) {  
170.             bytes = write(output_fd, buffer, BUF_SIZE);  
171.             bp = 0;  
172.         }  
173.         // mr的输入缓冲区用完  
174.         if (mr->idx == mr->length) {  
175.             snprintf(filename, 20, "%s%d.dat", input_prefix, ls[0]+n_merge*K);  
176.             if (base) {  
177.                 mr->buf[mr->idx] = MAX_INT;  
178.                 live_runs--;  
179.             } else {  
180.                 int fd = open(filename, O_RDONLY);  
181.                 lseek(fd, mr->offset, SEEK_SET);  
182.                 bytes = read(fd, mr->buf, BUF_SIZE);  
183.                 close(fd);  
184.                 if (bytes == 0) {  
185.                     mr->buf[mr->idx] = MAX_INT;  
186.                     live_runs--;  
187.                 }  
188.                 else {  
189.                     mr->length = bytes/sizeof(int);  
190.                     mr->offset += bytes;  
191.                     mr->idx = 0;  
192.                 }  
193.             }  
194.         }  
195.         adjust(runs, num_runs, ls[0]);  
196.     }  
197.     bytes = write(output_fd, buffer, bp*4);  
198.     if (bytes != bp*4) {  
199.         printf("!!!!!! Write Error !!!!!!!!!\n");  
200.         exit(0);  
201.     }  
202.     close(output_fd);  
203. }  
204.   
205. long get_time_usecs()  
206. {  
207.     struct timeval time;  
208.     struct timezone tz;  
209.     memset(&tz, '\0', sizeof(struct timezone));  
210.     gettimeofday(&time, &tz);  
211.     long usecs = time.tv_sec*1000000 + time.tv_usec;  
212.   
213.     return usecs;  
214. }  
215.   
216. void swap(int *p, int *q)  
217. {  
218.     int     tmp;  
219.   
220.     tmp = *p;  
221.     *p = *q;  
222.     *q = tmp;  
223. }  
224.   
225. int partition(int *a, int s, int t)  
226. {  
227.     int     i, j;   /* i用来遍历a[s]...a[t-1], j指向大于x部分的第一个元素 */  
228.   
229.     for (i = j = s; i < t; i++) {  
230.         if (a[i] < a[t]) {  
231.             swap(a+i, a+j);  
232.             j++;  
233.         }  
234.     }  
235.     swap(a+j, a+t);  
236.   
237.     return j;  
238. }  
239.   
240. void quick_sort(int *a, int s, int t)  
241. {  
242.     int     p;  
243.   
244.     if (s < t) {  
245.         p = partition(a, s, t);  
246.         quick_sort(a, s, p-1);  
247.         quick_sort(a, p+1, t);  
248.     }  
249. }  
250.   
251. void adjust(run_t ** runs, int n, int s)  
252. {  
253.     int t, tmp;  
254.   
255.     t = (s+n)/2;  
256.     while (t > 0) {  
257.         if (s == -1) {  
258.             break;  
259.         }  
260.         if (ls[t] == -1 || runs[s]->buf[runs[s]->idx] > runs[ls[t]]->buf[runs[ls[t]]->idx]) {  
261.             tmp = s;  
262.             s = ls[t];  
263.             ls[t] = tmp;  
264.         }  
265.         t >>= 1;  
266.     }  
267.     ls[0] = s;  
268. }  
269.   
270. void create_loser_tree(run_t **runs, int n)  
271. {  
272.     int     i;  
273.   
274.     for (i = 0; i < n; i++) {  
275.         ls[i] = -1;  
276.     }  
277.     for (i = n-1; i >= 0; i--) {  
278.         adjust(runs, n, i);  
279.     }  
280. }  
281.   
282. void usage()  
283. {  
284.     printf("sort <filename> <K-ways>\n");  
285.     printf("\tfilename: filename of file to be sorted\n");  
286.     printf("\tK-ways: how many ways to merge\n");  
287.     exit(1);  
288. }  

 

 

编译运行

gcc sort.c -o sort -g

./sort random.dat 64

以64路平衡归并对random.dat内的数据进行外部排序。在I5处理器，4G内存的硬件环境下，实验结果如下

文件大小    耗时

128M        14.72 秒

256M        30.89 秒

512M        71.65 秒

1G             169.18秒

 

读取二进制文件，查看排序结

[cpp] view plain copy

1. #include <assert.h>  
2. #include <fcntl.h>  
3. #include <stdio.h>  
4. #include <stdlib.h>  
5. #include <string.h>  
6. #include <unistd.h>  
7.   
8. #include <sys/time.h>  
9. #include <sys/types.h>  
10. #include <sys/stat.h>  
11.   
12. int main(int argc, char **argv)  
13. {  
14.     char *filename = argv[1];  
15.     int *buffer = (int *)malloc(1<<20);  
16.     struct stat     sbuf;  
17.     int rv, data_size, i, bytes, fd;  
18.   
19.     fd = open(filename, O_RDONLY);  
20.     if (fd < 0) {  
21.         printf("%s not found!\n", filename);  
22.         exit(0);  
23.     }  
24.     rv = fstat(fd, &sbuf);  
25.     data_size = sbuf.st_size;  
26.   
27.     bytes = read(fd, buffer, data_size);  
28.     for (i = 0; i < bytes/4; i++) {  
29.         printf("%d ", buffer[i]);  
30.         if ((i+1) % 10 == 0) {  
31.             printf("\n");  
32.         }  
33.     }  
34.     printf("\n");  
35.     close(fd);  
36.     free(buffer);  
37.     return 0;  
38. }