外排序的摸索
? ?今天晚上的目标就是实现一个外排序的算法，最近几天多多少少的看了点这方面的文章，还有一些实现，之前对这个概念十分不清晰，其实现在想来，外排序的操作文件，其实和操作内存一样，只不过它的速度实在是太慢了。但在代码上几乎没有区别，把内存上的定义数据，转变成对文件的读入读出。
? ?其实现在还在继续研究中，打算先把完成的一部分贴出来，然后全部完成后，再拿出完整版，这样有个思考的过程。不过说时候现在完成的这部分我都不好意思拿出来。。。还是皮厚点吧。。上代码
? ?这个函数是分割文件的，把一个大文件分割成N个小文件，从大文件中读取的数据量需指定，还要分割成的碎片文件的大小，当然得到的碎片文件是通过快排排序好的（随机法改进的快排）。
?
?
int dataCut(char* filename,int numSize,int totalSize){int i;int cutnum = ceil((double)totalSize/numSize);//计算切割成多少个文件int curNum=1;int numleft=totalSize%numSize;//计算最后一个文件的数字cout<<cutnum<<" "<<numleft<<endl;ifstream input(filename);//输入流int* a=(int*)malloc(numSize*sizeof(int));while(1){ofstream output(getFileName(curNum,0));if(curNum==cutnum) { for(i=0;i<numleft;++i) { input>>a[i]; } quickSort(a,0,numleft-1); for(i=0;i<numleft;++i) { output<<a[i]<<" "; output.flush(); } output.close(); break; } for(i=0;i<numSize;++i) { input>>a[i]; } quickSort(a,0,numSize-1); for(i=0;i<numSize;++i) { output<<a[i]<<" "; output.flush(); } output.close(); curNum++;} input.close(); free(a); return cutnum;//返回文件数目}??
?快排的代码其实之前的日志里有，不过这个版本是修改过的，也贴出来。
?
int qsort_partion(int* a,int start,int end{int i=start,j=end;int judge = a[ end ];//选择最后一个数字作为比较值while(1){while( a[i] < judge ) ++i;--j;while( a[j] > judge ) --j;if( i >= j ) break;else{swap(a[i],a[j]);//交换需要调整的值++i;}}swap(a[end],a[i]);//将当前值于比较值对换，然后就完成以i 为界，一边的值都小于a[i],另一边都大于a[i]return i;}int rand_partion(int *a,int start,int end)//这个就是修改的部分{int t= rand()%(end-start);swap(a[start+t],a[end]);return qsort_partion(a,start,end);}void quickSort(int* a, int start ,int end){if(start <end ){int q = rand_partion(a,start,end);quickSort(a,start,q-1);quickSort(a,q+1,end);}}?
? ?外排序的核心部分就是用归并排序合并得到最后的目标，我还没完成，但实现了一个合并两个文件的，思想上可以渐进吧，正在努力中，先把合并两个的贴出来。。这里默认每块文件都含有1024个整数，实际情况当然没这么简单。。
?
ifstream input1(".//data1.txt"); ifstream input2(".//data2.txt"); ofstream output(".//dataDst.txt"); int a[1024]; int b[1024]; for(i=0;i<1024;++i) { input1>>a[i]; input2>>b[i]; } int j,k; j=0;k=0; while(k<1024 && j<1024) { if( a[j]>b[k] ) { output<<b[k]<<" "; ++k; } else { output<<a[j]<<" "; ++j; } } input1.close(); input2.close(); output.close();? 然后继续努力！！
?
思路卡壳了，还是打会儿星际睡觉吧。。改日继续
?
11.26 早上又写了一会儿，发现卡壳的位置依旧没变，后来决定先看看别人怎么做的，参考了下JULY的博客，发现他里面的做法效率其实不高，仅仅维护一个整形数组保持每个文件的第一个数字，然后找到最小的一个写入目标文件，再从文件中更新下一个。这样需要频繁的读取文件，速度会比较慢。我自己的目标是实现一个 一次从每个文件读取一个小片段（加起来低于内存限制），然后在对各个小片段进行处理，这样的效率会高些。但我还想同时实现自动化的完成从目标文件到最终文件的排序。。然后这些组合一起导致了无比的复杂性。。。反正我是扛不住了，所以昨天到现在除了上面的成果外，无任何进展。
? ? 所以现在决定，还是一步步来把，写代码也需要循序渐进，从简单到复杂嘛，今天先完成July博客的那种做法（N个小文件一次合并成一个目标文件），然后继续摸索高级的（N个文件多次合并成一个目标文件）。
? ? 因为需要考虑效率问题，所以先贴一个计算时间的简单函数，C++真是很尴尬很多比较重要的东西都要没有，都需要用C版本的或者用系统的库，比如线程UI，还有时间统计。。。伤不起啊，用C的写了一个。
?
void showTimeUsed(time_t start){time_t end=time(NULL);cout<<"Use:"<<(end-start)<<"s"<<endl;}? ?在需要打印时间的地方调用，在计时开始的地方，需要有这句配合
time_t start=time(NULL);
?
? 11.26晚上，初步完成了K路合并为一路，测试了1000000数据量，大约运行时间3S。
? 现在将把各个部分的内容一一列出。
?
?
char* getFileName(int num,int times){ char* filename=(char*)malloc(sizeof(char)*50); sprintf(filename,".//data%d-%d.txt",num,times); return filename;}void dataInit()//生成数据{int i;ofstream file(".//data.txt"); if(!file.is_open()) { cout<<"NO File!"<<endl; } srand((int)time(0));for(i=0;i<1000000;++i)//生成的整数数量{file << rand()%1000000<<" ";//控制数据的范围}file.close();}?? getFileName是用来获取随机生成文件的名字，通过两个参数方便了解情况，第一个是索引，第二个是处理的次序。比如第一次DataCut切割完产生的文件就是data1-0.txt data2-0.txt以此类推。
? ?dataInit()是用来随机生成测试数据的，数据量和数据范围可以指定。。手动指定。。恩。
?
? ?然后就是重点了，K路合并，已经后面一个日志里说到的，封转从文件读取操作的AutoBuf类，就是尽量一次读入一块而不是每次都从文件里直接获取，这样的处理可以很大的提高速度（文件读入是系统的瓶颈）。代码如下。先是一些基本的定义。
?
?
const int K=1000;// 定义合并的K值const int memoryLimit=100000000;//单位Byteconst int bufSize=memoryLimit/K/sizeof(int);//每一路可以开辟的缓存大小 即每一组缓存最多有 bufSize 个整型?
然后AutoBuf类
?
struct AutoBuf{int cur;//当前数据的索引int* buf;//数据缓存int size;//缓存区现有数据大小bool isLast;//是否已经从文件读完ifstream in;AutoBuf(){cur=0;size=0;isLast=false;buf=(int*)malloc(sizeof(int)*bufSize);}~AutoBuf(){in.close();free(buf);}void AttachFile(const char* filename){in.open(filename);}int read()//从指定文件中读取{if( !in.is_open() ){cout<<"file not open!"<<endl;return -1;}int i=0;while( i<bufSize && in>>buf[i] ){++i;}size=i;//buf中有多少个数据cout<<"read size:"<<size<<endl;cur=0;if( size < bufSize ) isLast=true;return size;}int autoGet(){int tmp=buf[cur++];if( cur>=size ){if( isLast ){//已经没数据了return -1;}else//还有，可以继续读{if( read()==-1 )//读取失败{return -1;}}}return tmp;}};??? ?这个类其实就是一个中间作用，减少文件读写的次数，基本一次读入可以用好久了。本来还想写一个 减少文件写入的类，发现相似点太多了，改进也挺容易，就先留着这个问题。
?
? ? K路合并代码，目前只能处理一次合并的，递归多次合并即如何自动化完成全套流程，后面再想。
?
?
void K_merge(int K,const char* targetFile)//K路归并成一路{ int i,j; int min;//记录遍历过程的最小值 bool hasNext[K]; int Knum[K]; ofstream out( targetFile );//存入改文件中 AutoBuf buf[K]; for(i=0;i<K;++i) { buf[i].AttachFile(getFileName(i+1,0));//将自动缓存和文件绑定 buf[i].read();// 读取文件的其中一块 hasNext[i]=true;//设置文件未读完 Knum[i]=buf[i].autoGet();//获取第一个数字 } while(true) { j=0; while(j<K && !hasNext[j] ) ++j;//跳过已经读完的部分 if(j>=K) break;//已经读完了,退出 min=Knum[j]; if(min == -1) break; for(i=j+1;i<K;++i)//需找K组中的最小值 { if( hasNext[i] && Knum[i] < min ) { min=Knum[i];//记录最小值 j=i;//记录最小值所在的索引 } }// cout<<min<<endl; out<<min<<" "<<flush;//写入目标文件 Knum[j]=buf[j].autoGet(); //继续读入一个整形 if( Knum[j] == -1 ) //读入-1说明已经结束了 hasNext[j]=false; } out.close();}?? ?out<<min<<" ";这里可以添加所说的缓冲类，来减少写文件的次数，先把排序好的数字收集在缓冲中，然后达到一定数量后一次性写入，这样就完美了。。
? ?最后补充main函数，这样一个基本的版本就成型了。
?
?
int main(){ srand((int)time(0)); time_t start=time(NULL); dataInit(); int k=dataCut(".//data.txt",1024,1000000); K_merge(k,".//dst.txt"); showTimeUsed(start); return 0;}?? ?好吧，我承认我最后虐待了下电脑，生成了1亿个数字的文件，然后切割排序，但只归并一百路，用了93S。。
? ?最后检查数据的时候发现几个文件间歇性的出现乱码，想到下午那个问题，一查，果然是处理文件流的时候忘记flush了，吃一见长一智啊。这次加上就完全没问题了。
?
? ?11.26最终测试后发现，有很多Bug，比较严重的是，结果中居然出现-1.这个问题比较严重，然后就是效率偏低，直接归并法果然复杂度很高，数据量大的情况会越发明显。得引入其他的处理方法，比如败者树或者最小堆，今天就到这里了，改日慢慢改进。
?
? ?11.27日，实现使用维护K大的最小堆，来进行归并。
? ?补充一部分维护堆的代码，还有修改K_merge函数。
?
?
struct heapNode{AutoBuf* data;//数据源int Knum;//当前读入数值};typedef heapNode* Point;void adjust_heap(Point* a , int i)//调节该点 OK{Point tmp=a[i];while( i > 1 ){if( a[i>>1]->Knum > tmp->Knum){a[i]=a[i>>1];i>>=1;}else{break;}} a[i]=tmp;}void make_heap( Point* heap , int size )// OK{int i;for(i=1;i<=size;i++){adjust_heap(heap, i);}}void adjustDown_heap(Point* a ,int size){Point tmp = a[1];int i=1; while( i<= size )//有孩子 { if( 2*i > size ) break;//越界判断！！！！！ if( (2*i+1) > size)//只有一个孩子 { if( tmp->Knum < a[2*i]->Knum ) { break; } else { a[i]=a[2*i]; i=2*i; continue; } } if( tmp->Knum < a[2*i]->Knum && tmp->Knum < a[(2*i+1)]->Knum )//两个孩子 { break; } else//选择和较小的孩子节点替换 { if( a[2*i]->Knum < a[(2*i+1)]->Knum ) { a[i]=a[2*i] ; i*=2; } else { a[i]=a[(2*i+1)]; i=2*i+1; } } } a[i] = tmp;}void K_merge(int K,const char* targetFile)//K路归并成一路{ int i; int size = K; Point node[K+1];//这是一个指向heapNode对象的指针数组 Point tmp; for(i=0;i<=K;++i)//必须分配heapNode 对象的空间 node[i]=new heapNode; AutoBuf buf[K+1]; ofstream out( targetFile );//存入改文件中 for(i=1;i<=K;++i) { node[i]->data = buf+i; node[i]->data->AttachFile(getFileName(i,0));//将自动缓存和文件绑定 node[i]->data->read();// 读取文件的其中一块 node[i]->Knum=node[i]->data->autoGet();//获取第一个数字 } make_heap(node,size); while( size > 0 ) { out<<node[1]->Knum<<" "<<flush;//输出堆顶// cout<<node[1]->Knum<<" "; node[1]->Knum=node[1]->data->autoGet(); if( node[1]->Knum !=-1) { adjustDown_heap(node ,size); } else//堆大小减1，同时将其塞入废弃区 { tmp=node[1]; node[1]=node[size]; node[size]=tmp; --size; adjustDown_heap(node ,size); } } for(i=1;i<=K;++i) delete node[i]; out.close();}??? ?经测试发现这种方法比以前的快些，尤其是K的值比较大的时候，但如果K值很小，或者读写的文件块很大的时候，效率一般。相对朴素算法只是稍有提高而已。
?
? ?继续修改，因为感觉效率实在太差了，所以做了进一步的尝试，由于最终结果是得到一个数字写入一个，所以我觉得这里会有瓶颈，还是决定添加一个作为中间缓存的类AutoWriter，这个和AutoBuf的相当，只是负责写入端，没想到加入以后修改完，效果特别好。一下子把10的7次方数据的排序，从57S降到了34S，降低了23S，几乎是一大半的时间啊，原来文件读写的速度如此之慢。仅仅是块读取和一次一次读就差距这么大。
? ?先上代码，K_merge又改了。。这篇文章显得很乱，但正显示了一次一次的思考过程。
?
?
? ?首先是AutoWriter类
?
?
const int W_BUF=4096;//存满W_BUF个整数后写入文件struct AutoWriter{int *buf;int size;//缓存中的整形数ofstream out;AutoWriter(){size=0;buf=new int[W_BUF];}~AutoWriter(){Flush();out.close();delete[] buf;}void AttachFile(const char * filename){out.open(filename);}void put(int num){//cout<<num<<" ";if( size < W_BUF ){buf[size++]=num;}else{ Flush(); buf[size++]=num;}}void Flush(){for(int i=0;i<W_BUF;++i)//输出文件{out<<buf[i]<<" ";size=0;}out.flush();}};?? ?然后是K_merge
?
?
void K_merge(int K,const char* targetFile)//K路归并成一路{ int i; int size = K; Point node[K+1];//这是一个指向heapNode对象的指针数组 Point tmp; for(i=0;i<=K;++i)//必须分配heapNode 对象的空间 node[i]=new heapNode; AutoBuf buf[K+1]; AutoWriter w; w.AttachFile(targetFile); for(i=1;i<=K;++i) { node[i]->data = buf+i; node[i]->data->AttachFile(getFileName(i,0));//将自动缓存和文件绑定 node[i]->data->read();// 读取文件的其中一块 node[i]->Knum=node[i]->data->autoGet();//获取第一个数字 } make_heap(node,size); while( size > 0 ) { w.put(node[1]->Knum); node[1]->Knum=node[1]->data->autoGet(); if( node[1]->Knum !=-1) { adjustDown_heap(node ,size); } else//堆大小减1，同时将其塞入废弃区 { tmp=node[1]; node[1]=node[size]; node[size]=tmp; --size; adjustDown_heap(node ,size); } } for(i=1;i<=K;++i) delete node[i];}?
? ?现在就差最后两个尝试了，败者树可能比最小堆效率好些，然后就是能一次性的自动化完成多次K路归并。
? ?今天先到这里了。。
?
11.29 在摸索了一段时间后，终于着手写败者树实现的外部排序了。开始对败者树理解不到位，还对于完全二叉树的理解也不够，其实败者树说简单点就是：一堆记录比赛结果的节点（K）个，第0个位置记录的是最后的胜利者，每个节点的值是该节点子节点比赛的失败者。然后是K个选手，所以这个树的结构其实就是ls[0-k] players[0-k],然后只要你知道某个选手的索引，如s,那么他的父亲节点（即比赛结果记录的败者）的索引就是（2+k）/2 。每次数据更新，只需要和父亲节点记录的索引的选手比一次 ，然后记录败者，胜者沿树上升，直到根节点，这就是一轮调整。而调整的过程使用于建树，只要初始化一个最小值，然后对每一个选手节点进行调整就好，但不一定总知道所排序的数字的最小值和最大值嘛，那就随便找一个不冲突的作为该节点未比赛的记录，只要遇到这个值，就默认为胜者就成了。
?
这部分代码不贴出来了，占空间，就放在我的代码里。
?