外排序的摸索

   今天晚上的目标就是实现一个外排序的算法，最近几天多多少少的看了点这方面的文章，还有一些实现，之前对这个概念十分不清晰，其实现在想来，外排序的操作文件，其实和操作内存一样，只不过它的速度实在是太慢了。但在代码上几乎没有区别，把内存上的定义数据，转变成对文件的读入读出。

   其实现在还在继续研究中，打算先把完成的一部分贴出来，然后全部完成后，再拿出完整版，这样有个思考的过程。不过说时候现在完成的这部分我都不好意思拿出来。。。还是皮厚点吧。。上代码

   这个函数是分割文件的，把一个大文件分割成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;//单位Byte
const 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 。每次数据更新，只需要和父亲节点记录的索引的选手比一次 ，然后记录败者，胜者沿树上升，直到根节点，这就是一轮调整。而调整的过程使用于建树，只要初始化一个最小值，然后对每一个选手节点进行调整就好，但不一定总知道所排序的数字的最小值和最大值嘛，那就随便找一个不冲突的作为该节点未比赛的记录，只要遇到这个值，就默认为胜者就成了。

 

这部分代码不贴出来了，占空间，就放在我的代码里。