定義問題

外部排序指的是大檔案的排序，即待排序的記錄存儲在外存儲器上，待排序的檔案無法一次裝入記憶體，需要在記憶體和外部存儲器之間進行多次資料交換，以達到排序整個檔案的目的。外部排序最常用的算法是多路歸并排序，即将原檔案分解成多個能夠一次性裝入記憶體的部分，分别把每一部分調入記憶體完成排序。然後，對已經排序的子檔案進行多路歸并排序。

處理過程

（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所示。

2. b3 PK b0，b0勝b3負，内部結點ls[2]的值為0；

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

#include <assert.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>
#define MAX_INT ~(1<<31)
#define MIN_INT 1<<31
//#define DEBUG
#ifdef DEBUG
#define debug(...) debug( __VA_ARGS__)
#else
#define debug(...)
#endif
#define MAX_WAYS 100
typedef struct run_t {
int *buf; /* 輸入緩沖區 */
int length; /* 緩沖區目前有多少個數 */
int offset; /* 緩沖區讀到了檔案的哪個位置 */
int idx; /* 緩沖區的指針 */
} run_t;
static unsigned int K; /* K路合并 */
static unsigned int BUF_PAGES; /* 緩沖區有多少個page */
static unsigned int PAGE_SIZE; /* page的大小 */
static unsigned int BUF_SIZE; /* 緩沖區的大小, BUF_SIZE = BUF_PAGES*PAGE_SIZE */
static int *buffer; /* 輸出緩沖區 */
static char input_prefix[] = "foo_";
static char output_prefix[] = "bar_";
static int ls[MAX_WAYS]; /* loser tree */
void swap(int *p, int *q);
int partition(int *a, int s, int t);
void quick_sort(int *a, int s, int t);
void adjust(run_t ** runs, int n, int s);
void create_loser_tree(run_t **runs, int n);
long get_time_usecs();
void k_merge(run_t** runs, char* input_prefix, int num_runs, int base, int n_merge);
void usage();
int main(int argc, char **argv)
{
char filename[100];
unsigned int data_size;
unsigned int num_runs; /* 這輪疊代時有多少個歸并段 */
unsigned int num_merges; /* 這輪疊代後産生多少個歸并段 num_merges = num_runs/K */
unsigned int run_length; /* 歸并段的長度，指數級增長 */
unsigned int num_runs_in_merge; /* 一般每個merge由K個runs合并而來，但最後一個merge可能少于K個runs */
int fd, rv, i, j, bytes;
struct stat sbuf;
if (argc != 3) {
usage();
return 0;
}
long start_usecs = get_time_usecs();
strcpy(filename, argv[1]);
fd = open(filename, O_RDONLY);
if (fd < 0) {
printf("can't open file %s\n", filename);
exit(0);
rv = fstat(fd, &sbuf);
data_size = sbuf.st_size;
K = atoi(argv[2]);
PAGE_SIZE = 4096; /* page = 4KB */
BUF_PAGES = 32;
BUF_SIZE = PAGE_SIZE*BUF_PAGES;
num_runs = data_size / PAGE_SIZE; /* 初始時的歸并段數量，每個歸并段有4096 byte, 即1024個整數 */
buffer = (int *)malloc(BUF_SIZE);
run_length = 1;
run_t **runs = (run_t **)malloc(sizeof(run_t *)*(K+1));
for (i = 0; i < K; i++) {
runs[i] = (run_t *)malloc(sizeof(run_t));
runs[i]->buf = (int *)calloc(1, BUF_SIZE+4);
while (num_runs > 1) {
num_merges = num_runs / K;
int left_runs = num_runs % K;
if(left_runs > 0) num_merges++;
for (i = 0; i < num_merges; i++) {
num_runs_in_merge = K;
if ((i+1) == num_merges && left_runs > 0) {
num_runs_in_merge = left_runs;
}
int base = 0;
printf("Merge %d of %d,%d ways\n", i, num_merges, num_runs_in_merge);
for (j = 0; j < num_runs_in_merge; j++) {
if (run_length == 1) {
base = 1;
bytes = read(fd, runs[j]->buf, PAGE_SIZE);
runs[j]->length = bytes/sizeof(int);
quick_sort(runs[j]->buf, 0, runs[j]->length-1);
} else {
snprintf(filename, 20, "%s%d.dat", input_prefix, i*K+j);
int infd = open(filename, O_RDONLY);
bytes = read(infd, runs[j]->buf, BUF_SIZE);
close(infd);
}
runs[j]->idx = 0;
runs[j]->offset = bytes;
k_merge(runs, input_prefix, num_runs_in_merge, base, i);
}
strcpy(filename, output_prefix);
strcpy(output_prefix, input_prefix);
strcpy(input_prefix, filename);
run_length *= K;
num_runs = num_merges;
free(runs[i]->buf);
free(runs[i]);
free(runs);
free(buffer);
close(fd);
long end_usecs = get_time_usecs();
double secs = (double)(end_usecs - start_usecs) / (double)1000000;
printf("Sorting took %.02f seconds.\n", secs);
printf("sorting result saved in %s%d.dat.\n", input_prefix, 0);
return 0;
}
void k_merge(run_t** runs, char* input_prefix, int num_runs, int base, int n_merge)
int bp, bytes, output_fd;
int live_runs = num_runs;
run_t *mr;
char filename[20];
bp = 0;
create_loser_tree(runs, num_runs);
snprintf(filename, 100, "%s%d.dat", output_prefix, n_merge);
output_fd = open(filename, O_CREAT|O_WRONLY|O_TRUNC,
S_IRWXU|S_IRWXG);
if (output_fd < 0) {
printf("create file %s fail\n", filename);
while (live_runs > 0) {
mr = runs[ls[0]];
buffer[bp++] = mr->buf[mr->idx++];
// 輸出緩沖區已滿
if (bp*4 == BUF_SIZE) {
bytes = write(output_fd, buffer, BUF_SIZE);
bp = 0;
// mr的輸入緩沖區用完
if (mr->idx == mr->length) {
snprintf(filename, 20, "%s%d.dat", input_prefix, ls[0]+n_merge*K);
if (base) {
mr->buf[mr->idx] = MAX_INT;
live_runs--;
} else {
int fd = open(filename, O_RDONLY);
lseek(fd, mr->offset, SEEK_SET);
bytes = read(fd, mr->buf, BUF_SIZE);
close(fd);
if (bytes == 0) {
mr->buf[mr->idx] = MAX_INT;
live_runs--;
else {
mr->length = bytes/sizeof(int);
mr->offset += bytes;
mr->idx = 0;
adjust(runs, num_runs, ls[0]);
bytes = write(output_fd, buffer, bp*4);
if (bytes != bp*4) {
printf("!!!!!! Write Error !!!!!!!!!\n");
close(output_fd);
long get_time_usecs()
struct timeval time;
struct timezone tz;
memset(&tz, '\0', sizeof(struct timezone));
gettimeofday(&time, &tz);
long usecs = time.tv_sec*1000000 + time.tv_usec;
return usecs;
void swap(int *p, int *q)
int tmp;
tmp = *p;
*p = *q;
*q = tmp;
int partition(int *a, int s, int t)
int i, j; /* i用來周遊a[s]...a[t-1], j指向大于x部分的第一個元素 */
for (i = j = s; i < t; i++) {
if (a[i] < a[t]) {
swap(a+i, a+j);
j++;
swap(a+j, a+t);
return j;
void quick_sort(int *a, int s, int t)
int p;
if (s < t) {
p = partition(a, s, t);
quick_sort(a, s, p-1);
quick_sort(a, p+1, t);
void adjust(run_t ** runs, int n, int s)
int t, tmp;
t = (s+n)/2;
while (t > 0) {
if (s == -1) {
break;
if (ls[t] == -1 || runs[s]->buf[runs[s]->idx] > runs[ls[t]]->buf[runs[ls[t]]->idx]) {
tmp = s;
s = ls[t];
ls[t] = tmp;
t >>= 1;
ls[0] = s;
void create_loser_tree(run_t **runs, int n)
int i;
for (i = 0; i < n; i++) {
ls[i] = -1;
for (i = n-1; i >= 0; i--) {
adjust(runs, n, i);
void usage()
printf("sort <filename> <K-ways>\n");
printf("\tfilename: filename of file to be sorted\n");
printf("\tK-ways: how many ways to merge\n");
exit(1);

編譯運作

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秒

讀取二進制檔案，檢視排序結

char *filename = argv[1];
int *buffer = (int *)malloc(1<<20);
struct stat sbuf;
int rv, data_size, i, bytes, fd;
printf("%s not found!\n", filename);
bytes = read(fd, buffer, data_size);
for (i = 0; i < bytes/4; i++) {
printf("%d ", buffer[i]);
if ((i+1) % 10 == 0) {
printf("\n");
printf("\n");
}

常用的外部排序方法定義問題處理過程多路歸并排序算法以及敗者樹勝者樹與敗者樹敗者樹多路平衡歸并外部排序

定義問題

處理過程

多路歸并排序算法以及敗者樹

勝者樹與敗者樹

勝者樹

敗者樹

敗者樹多路平衡歸并外部排序

外部排序的基本思路

使用敗者樹加快合并排序

産生二進制測試資料

程式實作

編譯運作

讀取二進制檔案，檢視排序結

繼續閱讀

Codeforces 1417 D. Make Them Equal(思維+構造)

查找算法之二分查找查找算法之二分查找

查找算法學習之二分查找（Python版本）——BinarySearch

CQ V1.0分詞bates(基于雙數組tire樹)—應該是目前最快的中文分詞算法

Command Network(POJ 3164)---定根最小樹形圖模闆題題目描述輸入格式輸出格式輸入樣例輸出樣例分析源程式

開源低帶寬語音編解碼器

241 Different Ways to Add Parentheses（C代碼版）

【趨高機器視覺】機器視覺技術原了解析及解決方案

CSMA/CD1． CSMA/CD的概述2． CSMA 的工作原理3． CSMA/CD控制規程及特點4． CSMA/CD協定5． CSMA/CD的優點6．結束語

極大似然法(ML)與最大期望法(EM)

C++ 第十五周報告1--《冒泡法排序》

筆試面試題目：滑動視窗(二)

資料結構與算法（27）——排序（二）

Dijkstra--簡易版（最短路徑）

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

hdu7108哈希

常用的外部排序方法定義問題 處理過程 多路歸并排序算法以及敗者樹 勝者樹與敗者樹 敗者樹 多路平衡歸并外部排序

定義問題

處理過程

多路歸并排序算法以及敗者樹

勝者樹與敗者樹

勝者樹

敗者樹

外部排序的基本思路

使用敗者樹加快合并排序

産生二進制測試資料

程式實作

編譯運作

讀取二進制檔案，檢視排序結

繼續閱讀

常用的外部排序方法定義問題處理過程多路歸并排序算法以及敗者樹勝者樹與敗者樹敗者樹多路平衡歸并外部排序