外排序,当然在我看来,外排序考验的是一个程序员的架构能力,而不仅仅局限于排序这个层次。
一:N路归并排序
1.概序
我们知道算法中有一种叫做分治思想,一个大问题我们可以采取分而治之,各个突破,当子问题解决了,大问题也就KO了,还有一点我们知道
内排序的归并排序是采用二路归并的,因为分治后有LogN层,每层两路归并需要N的时候,最后复杂度为NlogN,那么外排序我们可以将这个“二”
扩大到M,也就是将一个大文件分成M个小文件,每个小文件是有序的,然后对应在内存中我们开M个优先队列,每个队列从对应编号的文件中读取
TopN条记录,然后我们从M路队列中各取一个数字进入中转站队列,并将该数字打上队列编号标记,当从中转站出来的最小数字就是我们最后要排
序的数字之一,因为该数字打上了队列编号,所以方便我们通知对应的编号队列继续出数字进入中转站队列,可以看出中转站一直保存了M个记录,
当中转站中的所有数字都出队完毕,则外排序结束。如果大家有点蒙的话,我再配合一张图,相信大家就会一目了然,这考验的是我们的架构能力。
图中这里有个Batch容器,这个容器我是基于性能考虑的,当batch=n时,我们定时刷新到文件中,保证内存有足够的空间。
2.构建
<1> 生成数据
这个基本没什么好说的,采用随机数生成n条记录。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | #region 随机生成数据 /// <summary> /// 随机生成数据 ///<param name="max">执行生成的数据上线</param> /// </summary> public static void CreateData( int max) {
var sw = new StreamWriter(Environment.CurrentDirectory + "//demo.txt" );
for ( int i = 0; i < max; i++)
{
Thread.Sleep(2);
var rand = new Random(( int )DateTime.Now.Ticks).Next(0, int .MaxValue >> 3);
sw.WriteLine(rand);
}
sw.Close(); } #endregion |
<2> 切分数据
根据实际情况我们来决定到底要分成多少个小文件,并且小文件的数据必须是有序的,小文件的个数也对应这内存中有多少个优先队列。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | #region 将数据进行分份 /// <summary> /// 将数据进行分份 /// <param name="size">每页要显示的条数</param> /// </summary> public static int Split( int size) {
//文件总记录数
int totalCount = 0;
//每一份文件存放 size 条 记录
List< int > small = new List< int >();
var sr = new StreamReader((Environment.CurrentDirectory + "//demo.txt" ));
var pageSize = size;
int pageCount = 0;
int pageIndex = 0;
while ( true )
{
var line = sr.ReadLine();
if (! string .IsNullOrEmpty(line))
{
totalCount++;
//加入小集合中
small.Add(Convert.ToInt32(line));
//说明已经到达指定的 size 条数了
if (totalCount % pageSize == 0)
{
pageIndex = totalCount / pageSize;
small = small.OrderBy(i => i).Select(i => i).ToList();
File.WriteAllLines(Environment.CurrentDirectory + "//" + pageIndex + ".txt" , small.Select(i => i.ToString()));
small.Clear();
}
}
else
{
//说明已经读完了,将剩余的small记录写入到文件中
pageCount = ( int )Math.Ceiling(( double )totalCount / pageSize);
small = small.OrderBy(i => i).Select(i => i).ToList();
File.WriteAllLines(Environment.CurrentDirectory + "//" + pageCount + ".txt" , small.Select(i => i.ToString()));
break ;
}
}
return pageCount; } #endregion |
<3> 加入队列
我们知道内存队列存放的只是小文件的topN条记录,当内存队列为空时,我们需要再次从小文件中读取下一批的TopN条数据,然后放入中转站
继续进行比较。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 | #region 将数据加入指定编号队列
/// <summary>
/// 将数据加入指定编号队列
/// </summary>
/// <param name="i">队列编号</param>
/// <param name="skip">文件中跳过的条数</param>
/// <param name="list"></param>
/// <param name="top">需要每次读取的条数</param>
public static void AddQueue( int i, List<PriorityQueue< int ?>> list, ref int [] skip, int top = 100)
{
var result = File.ReadAllLines((Environment.CurrentDirectory + "//" + (i + 1) + ".txt" ))
.Skip(skip[i]).Take(top).Select(j => Convert.ToInt32(j));
//加入到集合中
foreach ( var item in result)
list[i].Eequeue( null , item);
//将个数累计到skip中,表示下次要跳过的记录数
skip[i] += result.Count();
}
#endregion |
<4> 测试
最后我们来测试一下:
数据量:short.MaxValue。
内存存放量:1200。
在这种场景下,我们决定每个文件放1000条,也就有33个小文件,也就有33个内存队列,每个队列取Top100条,Batch=500时刷新
硬盘,中转站存放33*2个数字(因为入中转站时打上了队列标记),最后内存活动最大总数为:sum=33*100+500+66=896<1200。
时间复杂度为N*logN。当然这个“阀值”,我们可以再仔细微调。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | public static void Main()
{
//生成2^15数据
CreateData( short .MaxValue);
//每个文件存放1000条
var pageSize = 1000;
//达到batchCount就刷新记录
var batchCount = 0;
//判断需要开启的队列
var pageCount = Split(pageSize);
//内存限制:1500条
List<PriorityQueue< int ?>> list = new List<PriorityQueue< int ?>>();
//定义一个队列中转器
PriorityQueue< int ?> queueControl = new PriorityQueue< int ?>();
//定义每个队列完成状态
bool [] complete = new bool [pageCount];
//队列读取文件时应该跳过的记录数
int [] skip = new int [pageCount];
//是否所有都完成了
int allcomplete = 0;
//定义 10 个队列
for ( int i = 0; i < pageCount; i++)
{
list.Add( new PriorityQueue< int ?>());
//i: 记录当前的队列编码
//list: 队列数据
//skip:跳过的条数
AddQueue(i, list, ref skip);
}
//初始化操作,从每个队列中取出一条记录,并且在入队的过程中
//记录该数据所属的 “队列编号”
for ( int i = 0; i < list.Count; i++)
{
var temp = list[i].Dequeue();
//i:队列编码,level:要排序的数据
queueControl.Eequeue(i, temp.level);
}
//默认500条写入一次文件
List< int > batch = new List< int >();
//记录下次应该从哪一个队列中提取数据
int nextIndex = 0;
while (queueControl.Count() > 0)
{
//从中转器中提取数据
var single = queueControl.Dequeue();
//记录下一个队列总应该出队的数据
nextIndex = single.t.Value;
var nextData = list[nextIndex].Dequeue();
//如果改对内弹出为null,则说明该队列已经,需要从nextIndex文件中读取数据
if (nextData == null )
{
//如果该队列没有全部读取完毕
if (!complete[nextIndex])
{
AddQueue(nextIndex, list, ref skip);
//如果从文件中读取还是没有,则说明改文件中已经没有数据了
if (list[nextIndex].Count() == 0)
{
complete[nextIndex] = true ;
allcomplete++;
}
else
{
nextData = list[nextIndex].Dequeue();
}
}
}
//如果弹出的数不为空,则将该数入中转站
if (nextData != null )
{
//将要出队的数据 转入 中转站
queueControl.Eequeue(nextIndex, nextData.level);
}
batch.Add(single.level);
//如果batch=500,或者所有的文件都已经读取完毕,此时我们要批量刷入数据
if (batch.Count == batchCount || allcomplete == pageCount)
{
var sw = new StreamWriter(Environment.CurrentDirectory + "//result.txt" , true );
foreach ( var item in batch)
{
sw.WriteLine(item);
}
sw.Close();
batch.Clear();
}
}
Console.WriteLine( "恭喜,外排序完毕!" );
Console.Read();
} |
总的代码:
1 using System; 2 using System.Collections.Generic; 3 using System.Linq; 4 using System.Text; 5 using System.Diagnostics; 6 using System.Threading; 7 using System.IO; 8 using System.Threading.Tasks; 9 10 namespace ConsoleApplication2 11 { 12 public class Program 13 { 14 public static void Main() 15 { 16 //生成2^15数据 17 CreateData(short.MaxValue); 18 19 //每个文件存放1000条 20 var pageSize = 1000; 21 22 //达到batchCount就刷新记录 23 var batchCount = 0; 24 25 //判断需要开启的队列 26 var pageCount = Split(pageSize); 27 28 //内存限制:1500条 29 List<PriorityQueue<int?>> list = new List<PriorityQueue<int?>>(); 30 31 //定义一个队列中转器 32 PriorityQueue<int?> queueControl = new PriorityQueue<int?>(); 33 34 //定义每个队列完成状态 35 bool[] complete = new bool[pageCount]; 36 37 //队列读取文件时应该跳过的记录数 38 int[] skip = new int[pageCount]; 39 40 //是否所有都完成了 41 int allcomplete = 0; 42 43 //定义 10 个队列 44 for (int i = 0; i < pageCount; i++) 45 { 46 list.Add(new PriorityQueue<int?>()); 47 48 //i: 记录当前的队列编码 49 //list: 队列数据 50 //skip:跳过的条数 51 AddQueue(i, list, ref skip); 52 } 53 54 //初始化操作,从每个队列中取出一条记录,并且在入队的过程中 55 //记录该数据所属的 “队列编号” 56 for (int i = 0; i < list.Count; i++) 57 { 58 var temp = list[i].Dequeue(); 59 60 //i:队列编码,level:要排序的数据 61 queueControl.Eequeue(i, temp.level); 62 } 63 64 //默认500条写入一次文件 65 List<int> batch = new List<int>(); 66 67 //记录下次应该从哪一个队列中提取数据 68 int nextIndex = 0; 69 70 while (queueControl.Count() > 0) 71 { 72 //从中转器中提取数据 73 var single = queueControl.Dequeue(); 74 75 //记录下一个队列总应该出队的数据 76 nextIndex = single.t.Value; 77 78 var nextData = list[nextIndex].Dequeue(); 79 80 //如果改对内弹出为null,则说明该队列已经,需要从nextIndex文件中读取数据 81 if (nextData == null) 82 { 83 //如果该队列没有全部读取完毕 84 if (!complete[nextIndex]) 85 { 86 AddQueue(nextIndex, list, ref skip); 87 88 //如果从文件中读取还是没有,则说明改文件中已经没有数据了 89 if (list[nextIndex].Count() == 0) 90 { 91 complete[nextIndex] = true; 92 allcomplete++; 93 } 94 else 95 { 96 nextData = list[nextIndex].Dequeue(); 97 } 98 } 99 }100 101 //如果弹出的数不为空,则将该数入中转站102 if (nextData != null)103 {104 //将要出队的数据 转入 中转站105 queueControl.Eequeue(nextIndex, nextData.level);106 }107 108 batch.Add(single.level);109 110 //如果batch=500,或者所有的文件都已经读取完毕,此时我们要批量刷入数据111 if (batch.Count == batchCount || allcomplete == pageCount)112 {113 var sw = new StreamWriter(Environment.CurrentDirectory + "//result.txt", true);114 115 foreach (var item in batch)116 {117 sw.WriteLine(item);118 }119 120 sw.Close();121 122 batch.Clear();123 }124 }125 126 Console.WriteLine("恭喜,外排序完毕!");127 Console.Read();128 }129 130 #region 将数据加入指定编号队列131 /// <summary>132 /// 将数据加入指定编号队列133 /// </summary>134 /// <param name="i">队列编号</param>135 /// <param name="skip">文件中跳过的条数</param>136 /// <param name="list"></param>137 /// <param name="top">需要每次读取的条数</param>138 public static void AddQueue(int i, List<PriorityQueue<int?>> list, ref int[] skip, int top = 100)139 {140 var result = File.ReadAllLines((Environment.CurrentDirectory + "//" + (i + 1) + ".txt"))141 .Skip(skip[i]).Take(top).Select(j => Convert.ToInt32(j));142 143 //加入到集合中144 foreach (var item in result)145 list[i].Eequeue(null, item);146 147 //将个数累计到skip中,表示下次要跳过的记录数148 skip[i] += result.Count();149 }150 #endregion151 152 #region 随机生成数据153 /// <summary>154 /// 随机生成数据155 ///<param name="max">执行生成的数据上线</param>156 /// </summary>157 public static void CreateData(int max)158 {159 var sw = new StreamWriter(Environment.CurrentDirectory + "//demo.txt");160 161 for (int i = 0; i < max; i++)162 {163 Thread.Sleep(2);164 var rand = new Random((int)DateTime.Now.Ticks).Next(0, int.MaxValue >> 3);165 166 sw.WriteLine(rand);167 }168 sw.Close();169 }170 #endregion171 172 #region 将数据进行分份173 /// <summary>174 /// 将数据进行分份175 /// <param name="size">每页要显示的条数</param>176 /// </summary>177 public static int Split(int size)178 {179 //文件总记录数180 int totalCount = 0;181 182 //每一份文件存放 size 条 记录183 List<int> small = new List<int>();184 185 var sr = new StreamReader((Environment.CurrentDirectory + "//demo.txt"));186 187 var pageSize = size;188 189 int pageCount = 0;190 191 int pageIndex = 0;192 193 while (true)194 {195 var line = sr.ReadLine();196 197 if (!string.IsNullOrEmpty(line))198 {199 totalCount++;200 201 //加入小集合中202 small.Add(Convert.ToInt32(line));203 204 //说明已经到达指定的 size 条数了205 if (totalCount % pageSize == 0)206 {207 pageIndex = totalCount / pageSize;208 209 small = small.OrderBy(i => i).Select(i => i).ToList();210 211 File.WriteAllLines(Environment.CurrentDirectory + "//" + pageIndex + ".txt", small.Select(i => i.ToString()));212 213 small.Clear();214 }215 }216 else217 {218 //说明已经读完了,将剩余的small记录写入到文件中219 pageCount = (int)Math.Ceiling((double)totalCount / pageSize);220 221 small = small.OrderBy(i => i).Select(i => i).ToList();222 223 File.WriteAllLines(Environment.CurrentDirectory + "//" + pageCount + ".txt", small.Select(i => i.ToString()));224 225 break;226 }227 }228 229 return pageCount;230 }231 #endregion232 }233 }
优先队列:
1 using System; 2 using System.Collections.Generic; 3 using System.Linq; 4 using System.Text; 5 using System.Diagnostics; 6 using System.Threading; 7 using System.IO; 8 9 namespace ConsoleApplication2 10 { 11 public class PriorityQueue<T> 12 { 13 /// <summary> 14 /// 定义一个数组来存放节点 15 /// </summary> 16 private List<HeapNode> nodeList = new List<HeapNode>(); 17 18 #region 堆节点定义 19 /// <summary> 20 /// 堆节点定义 21 /// </summary> 22 public class HeapNode 23 { 24 /// <summary> 25 /// 实体数据 26 /// </summary> 27 public T t { get; set; } 28 29 /// <summary> 30 /// 优先级别 1-10个级别 (优先级别递增) 31 /// </summary> 32 public int level { get; set; } 33 34 public HeapNode(T t, int level) 35 { 36 this.t = t; 37 this.level = level; 38 } 39 40 public HeapNode() { } 41 } 42 #endregion 43 44 #region 添加操作 45 /// <summary> 46 /// 添加操作 47 /// </summary> 48 public void Eequeue(T t, int level = 1) 49 { 50 //将当前节点追加到堆尾 51 nodeList.Add(new HeapNode(t, level)); 52 53 //如果只有一个节点,则不需要进行筛操作 54 if (nodeList.Count == 1) 55 return; 56 57 //获取最后一个非叶子节点 58 int parent = nodeList.Count / 2 - 1; 59 60 //堆调整 61 UpHeapAdjust(nodeList, parent); 62 } 63 #endregion 64 65 #region 对堆进行上滤操作,使得满足堆性质 66 /// <summary> 67 /// 对堆进行上滤操作,使得满足堆性质 68 /// </summary> 69 /// <param name="nodeList"></param> 70 /// <param name="index">非叶子节点的之后指针(这里要注意:我们 71 /// 的筛操作时针对非叶节点的) 72 /// </param> 73 public void UpHeapAdjust(List<HeapNode> nodeList, int parent) 74 { 75 while (parent >= 0) 76 { 77 //当前index节点的左孩子 78 var left = 2 * parent + 1; 79 80 //当前index节点的右孩子 81 var right = left + 1; 82 83 //parent子节点中最大的孩子节点,方便于parent进行比较 84 //默认为left节点 85 var min = left; 86 87 //判断当前节点是否有右孩子 88 if (right < nodeList.Count) 89 { 90 //判断parent要比较的最大子节点 91 min = nodeList[left].level < nodeList[right].level ? left : right; 92 } 93 94 //如果parent节点大于它的某个子节点的话,此时筛操作 95 if (nodeList[parent].level > nodeList[min].level) 96 { 97 //子节点和父节点进行交换操作 98 var temp = nodeList[parent]; 99 nodeList[parent] = nodeList[min];100 nodeList[min] = temp;101 102 //继续进行更上一层的过滤103 parent = (int)Math.Ceiling(parent / 2d) - 1;104 }105 else106 {107 break;108 }109 }110 }111 #endregion112 113 #region 优先队列的出队操作114 /// <summary>115 /// 优先队列的出队操作116 /// </summary>117 /// <returns></returns>118 public HeapNode Dequeue()119 {120 if (nodeList.Count == 0)121 return null;122 123 //出队列操作,弹出数据头元素124 var pop = nodeList[0];125 126 //用尾元素填充头元素127 nodeList[0] = nodeList[nodeList.Count - 1];128 129 //删除尾节点130 nodeList.RemoveAt(nodeList.Count - 1);131 132 //然后从根节点下滤堆133 DownHeapAdjust(nodeList, 0);134 135 return pop;136 }137 #endregion138 139 #region 对堆进行下滤操作,使得满足堆性质140 /// <summary>141 /// 对堆进行下滤操作,使得满足堆性质142 /// </summary>143 /// <param name="nodeList"></param>144 /// <param name="index">非叶子节点的之后指针(这里要注意:我们145 /// 的筛操作时针对非叶节点的)146 /// </param>147 public void DownHeapAdjust(List<HeapNode> nodeList, int parent)148 {149 while (2 * parent + 1 < nodeList.Count)150 {151 //当前index节点的左孩子152 var left = 2 * parent + 1;153 154 //当前index节点的右孩子155 var right = left + 1;156 157 //parent子节点中最大的孩子节点,方便于parent进行比较158 //默认为left节点159 var min = left;160 161 //判断当前节点是否有右孩子162 if (right < nodeList.Count)163 {164 //判断parent要比较的最大子节点165 min = nodeList[left].level < nodeList[right].level ? left : right;166 }167 168 //如果parent节点小于它的某个子节点的话,此时筛操作169 if (nodeList[parent].level > nodeList[min].level)170 {171 //子节点和父节点进行交换操作172 var temp = nodeList[parent];173 nodeList[parent] = nodeList[min];174 nodeList[min] = temp;175 176 //继续进行更下一层的过滤177 parent = min;178 }179 else180 {181 break;182 }183 }184 }185 #endregion186 187 #region 获取元素并下降到指定的level级别188 /// <summary>189 /// 获取元素并下降到指定的level级别190 /// </summary>191 /// <returns></returns>192 public HeapNode GetAndDownPriority(int level)193 {194 if (nodeList.Count == 0)195 return null;196 197 //获取头元素198 var pop = nodeList[0];199 200 //设置指定优先级(如果为 MinValue 则为 -- 操作)201 nodeList[0].level = level == int.MinValue ? --nodeList[0].level : level;202 203 //下滤堆204 DownHeapAdjust(nodeList, 0);205 206 return nodeList[0];207 }208 #endregion209 210 #region 获取元素并下降优先级211 /// <summary>212 /// 获取元素并下降优先级213 /// </summary>214 /// <returns></returns>215 public HeapNode GetAndDownPriority()216 {217 //下降一个优先级218 return GetAndDownPriority(int.MinValue);219 }220 #endregion221 222 #region 返回当前优先队列中的元素个数223 /// <summary>224 /// 返回当前优先队列中的元素个数225 /// </summary>226 /// <returns></returns>227 public int Count()228 {229 return nodeList.Count;230 }231 #endregion232 }233 }
联系客服