加载中...

第十八题 外排序


说到排序,大家第一反应基本上是内排序,是的,算法嘛,玩的就是内存,然而内存是有限制的,总有装不下的那一天,此时就可以来玩玩

外排序,当然在我看来,外排序考验的是一个程序员的架构能力,而不仅仅局限于排序这个层次。

 

一:N路归并排序

1.概序

    我们知道算法中有一种叫做分治思想,一个大问题我们可以采取分而治之,各个突破,当子问题解决了,大问题也就KO了,还有一点我们知道

内排序的归并排序是采用二路归并的,因为分治后有LogN层,每层两路归并需要N的时候,最后复杂度为NlogN,那么外排序我们可以将这个“二”

扩大到M,也就是将一个大文件分成M个小文件,每个小文件是有序的,然后对应在内存中我们开M个优先队列,每个队列从对应编号的文件中读取

TopN条记录,然后我们从M路队列中各取一个数字进入中转站队列,并将该数字打上队列编号标记,当从中转站出来的最小数字就是我们最后要排

序的数字之一,因为该数字打上了队列编号,所以方便我们通知对应的编号队列继续出数字进入中转站队列,可以看出中转站一直保存了M个记录,

当中转站中的所有数字都出队完毕,则外排序结束。如果大家有点蒙的话,我再配合一张图,相信大家就会一目了然,这考验的是我们的架构能力。

图中这里有个Batch容器,这个容器我是基于性能考虑的,当batch=n时,我们定时刷新到文件中,保证内存有足够的空间。

 

2.构建

<1> 生成数据

   这个基本没什么好说的,采用随机数生成n条记录。

  1. #region 随机生成数据
  2. /// <summary>
  3. /// 随机生成数据
  4. ///<param name="max">执行生成的数据上线</param>
  5. /// </summary>
  6. public static void CreateData(int max)
  7. {
  8. var sw = new StreamWriter(Environment.CurrentDirectory + "//demo.txt");
  9. for (int i = 0; i < max; i++)
  10. {
  11. Thread.Sleep(2);
  12. var rand = new Random((int)DateTime.Now.Ticks).Next(0, int.MaxValue >> 3);
  13. sw.WriteLine(rand);
  14. }
  15. sw.Close();
  16. }
  17. #endregion

<2> 切分数据

     根据实际情况我们来决定到底要分成多少个小文件,并且小文件的数据必须是有序的,小文件的个数也对应这内存中有多少个优先队列。

  1. #region 将数据进行分份
  2. /// <summary>
  3. /// 将数据进行分份
  4. /// <param name="size">每页要显示的条数</param>
  5. /// </summary>
  6. public static int Split(int size)
  7. {
  8. //文件总记录数
  9. int totalCount = 0;
  10. //每一份文件存放 size 条 记录
  11. List<int> small = new List<int>();
  12. var sr = new StreamReader((Environment.CurrentDirectory + "//demo.txt"));
  13. var pageSize = size;
  14. int pageCount = 0;
  15. int pageIndex = 0;
  16. while (true)
  17. {
  18. var line = sr.ReadLine();
  19. if (!string.IsNullOrEmpty(line))
  20. {
  21. totalCount++;
  22. //加入小集合中
  23. small.Add(Convert.ToInt32(line));
  24. //说明已经到达指定的 size 条数了
  25. if (totalCount % pageSize == 0)
  26. {
  27. pageIndex = totalCount / pageSize;
  28. small = small.OrderBy(i => i).Select(i => i).ToList();
  29. File.WriteAllLines(Environment.CurrentDirectory + "//" + pageIndex + ".txt", small.Select(i => i.ToString()));
  30. small.Clear();
  31. }
  32. }
  33. else
  34. {
  35. //说明已经读完了,将剩余的small记录写入到文件中
  36. pageCount = (int)Math.Ceiling((double)totalCount / pageSize);
  37. small = small.OrderBy(i => i).Select(i => i).ToList();
  38. File.WriteAllLines(Environment.CurrentDirectory + "//" + pageCount + ".txt", small.Select(i => i.ToString()));
  39. break;
  40. }
  41. }
  42. return pageCount;
  43. }
  44. #endregion

<3> 加入队列

    我们知道内存队列存放的只是小文件的topN条记录,当内存队列为空时,我们需要再次从小文件中读取下一批的TopN条数据,然后放入中转站

继续进行比较。

  1. #region 将数据加入指定编号队列
  2. /// <summary>
  3. /// 将数据加入指定编号队列
  4. /// </summary>
  5. /// <param name="i">队列编号</param>
  6. /// <param name="skip">文件中跳过的条数</param>
  7. /// <param name="list"></param>
  8. /// <param name="top">需要每次读取的条数</param>
  9. public static void AddQueue(int i, List<PriorityQueue<int?>> list, ref int[] skip, int top = 100)
  10. {
  11. var result = File.ReadAllLines((Environment.CurrentDirectory + "//" + (i + 1) + ".txt"))
  12. .Skip(skip[i]).Take(top).Select(j => Convert.ToInt32(j));
  13. //加入到集合中
  14. foreach (var item in result)
  15. list[i].Eequeue(null, item);
  16. //将个数累计到skip中,表示下次要跳过的记录数
  17. skip[i] += result.Count();
  18. }
  19. #endregion

<4> 测试

 最后我们来测试一下:

 数据量:short.MaxValue。

 内存存放量:1200。

在这种场景下,我们决定每个文件放1000条,也就有33个小文件,也就有33个内存队列,每个队列取Top100条,Batch=500时刷新

硬盘,中转站存放33*2个数字(因为入中转站时打上了队列标记),最后内存活动最大总数为:sum=33*100+500+66=896<1200。

时间复杂度为N*logN。当然这个“阀值”,我们可以再仔细微调。

  1. public static void Main()
  2. {
  3. //生成2^15数据
  4. CreateData(short.MaxValue);
  5. //每个文件存放1000条
  6. var pageSize = 1000;
  7. //达到batchCount就刷新记录
  8. var batchCount = 0;
  9. //判断需要开启的队列
  10. var pageCount = Split(pageSize);
  11. //内存限制:1500条
  12. List<PriorityQueue<int?>> list = new List<PriorityQueue<int?>>();
  13. //定义一个队列中转器
  14. PriorityQueue<int?> queueControl = new PriorityQueue<int?>();
  15. //定义每个队列完成状态
  16. bool[] complete = new bool[pageCount];
  17. //队列读取文件时应该跳过的记录数
  18. int[] skip = new int[pageCount];
  19. //是否所有都完成了
  20. int allcomplete = 0;
  21. //定义 10 个队列
  22. for (int i = 0; i < pageCount; i++)
  23. {
  24. list.Add(new PriorityQueue<int?>());
  25. //i: 记录当前的队列编码
  26. //list: 队列数据
  27. //skip:跳过的条数
  28. AddQueue(i, list, ref skip);
  29. }
  30. //初始化操作,从每个队列中取出一条记录,并且在入队的过程中
  31. //记录该数据所属的 “队列编号”
  32. for (int i = 0; i < list.Count; i++)
  33. {
  34. var temp = list[i].Dequeue();
  35. //i:队列编码,level:要排序的数据
  36. queueControl.Eequeue(i, temp.level);
  37. }
  38. //默认500条写入一次文件
  39. List<int> batch = new List<int>();
  40. //记录下次应该从哪一个队列中提取数据
  41. int nextIndex = 0;
  42. while (queueControl.Count() > 0)
  43. {
  44. //从中转器中提取数据
  45. var single = queueControl.Dequeue();
  46. //记录下一个队列总应该出队的数据
  47. nextIndex = single.t.Value;
  48. var nextData = list[nextIndex].Dequeue();
  49. //如果改对内弹出为null,则说明该队列已经,需要从nextIndex文件中读取数据
  50. if (nextData == null)
  51. {
  52. //如果该队列没有全部读取完毕
  53. if (!complete[nextIndex])
  54. {
  55. AddQueue(nextIndex, list, ref skip);
  56. //如果从文件中读取还是没有,则说明改文件中已经没有数据了
  57. if (list[nextIndex].Count() == 0)
  58. {
  59. complete[nextIndex] = true;
  60. allcomplete++;
  61. }
  62. else
  63. {
  64. nextData = list[nextIndex].Dequeue();
  65. }
  66. }
  67. }
  68. //如果弹出的数不为空,则将该数入中转站
  69. if (nextData != null)
  70. {
  71. //将要出队的数据 转入 中转站
  72. queueControl.Eequeue(nextIndex, nextData.level);
  73. }
  74. batch.Add(single.level);
  75. //如果batch=500,或者所有的文件都已经读取完毕,此时我们要批量刷入数据
  76. if (batch.Count == batchCount || allcomplete == pageCount)
  77. {
  78. var sw = new StreamWriter(Environment.CurrentDirectory + "//result.txt", true);
  79. foreach (var item in batch)
  80. {
  81. sw.WriteLine(item);
  82. }
  83. sw.Close();
  84. batch.Clear();
  85. }
  86. }
  87. Console.WriteLine("恭喜,外排序完毕!");
  88. Console.Read();
  89. }

总的代码:

  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. #endregion
  151.  
  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. #endregion
  171.  
  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. else
  217. {
  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. #endregion
  232. }
  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. else
  106. {
  107. break;
  108. }
  109. }
  110. }
  111. #endregion
  112.  
  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. #endregion
  138.  
  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. else
  180. {
  181. break;
  182. }
  183. }
  184. }
  185. #endregion
  186.  
  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. #endregion
  209.  
  210. #region 获取元素并下降优先级
  211. /// <summary>
  212. /// 获取元素并下降优先级
  213. /// </summary>
  214. /// <returns></returns>
  215. public HeapNode GetAndDownPriority()
  216. {
  217. //下降一个优先级
  218. return GetAndDownPriority(int.MinValue);
  219. }
  220. #endregion
  221.  
  222. #region 返回当前优先队列中的元素个数
  223. /// <summary>
  224. /// 返回当前优先队列中的元素个数
  225. /// </summary>
  226. /// <returns></returns>
  227. public int Count()
  228. {
  229. return nodeList.Count;
  230. }
  231. #endregion
  232. }
  233. }


还没有评论.