首页 > 代码库 > 经典算法题每日演练——第十八题 外排序

经典算法题每日演练——第十八题 外排序

原文:经典算法题每日演练——第十八题 外排序

 

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

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

 

一:N路归并排序

1.概序

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

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

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

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

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

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

技术分享

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

 

2.构建

<1> 生成数据

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

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

        #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条数据,然后放入中转站

继续进行比较。

#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。当然这个“阀值”,我们可以再仔细微调。

  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 = http://www.mamicode.com/list[nextIndex].Dequeue();"//result.txt", true);                    foreach (var item in batch)                    {                        sw.WriteLine(item);                    }                    sw.Close();                    batch.Clear();                }            }            Console.WriteLine("恭喜,外排序完毕!");            Console.Read();        }

  

技术分享

总的代码:

技术分享View Code
  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 = http://www.mamicode.com/list[nextIndex].Dequeue();"//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 }

 

优先队列:

技术分享View Code
  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 }

 

经典算法题每日演练——第十八题 外排序