外排序－多路歸併
阿新 • • 發佈：2019-02-02
說到排序，大家第一反應基本上是內排序，是的，演算法嘛，玩的就是記憶體，然而記憶體是有限制的，總有裝不下的那一天，此時就可以來玩玩
外排序，當然在我看來，外排序考驗的是一個程式設計師的架構能力，而不僅僅侷限於排序這個層次。
一：N路歸併排序
1.概序
我們知道演算法中有一種叫做分治思想，一個大問題我們可以採取分而治之，各個突破，當子問題解決了，大問題也就KO了，還有一點我們知道內排序的歸併排序是採用二路歸併的，因為分治後有LogN層，每層兩路歸併需要N的時候，最後複雜度為NlogN，那麼外排序我們可以將這個“二”擴大到M，也就是將一個大檔案分成M個小檔案，每個小檔案是有序的，然後對應在記憶體中我們開M個優先佇列，每個佇列從對應編號的檔案中讀取
圖中這裡有個Batch容器，這個容器我是基於效能考慮的，當batch=n時，我們定時重新整理到檔案中，保證記憶體有足夠的空間。
2.構建
<1> 生成資料
這個基本沒什麼好說的，採用隨機數生成n條記錄。
<span style="font-size:18px;">#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</span>
<2> 切分資料
根據實際情況我們來決定到底要分成多少個小檔案，並且小檔案的資料必須是有序的，小檔案的個數也對應這記憶體中有多少個優先佇列。
<span style="font-size:18px;">#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</span><3> 加入佇列
我們知道記憶體佇列存放的只是小檔案的topN條記錄，當記憶體佇列為空時，我們需要再次從小檔案中讀取下一批的TopN條資料，然後放入中轉站
繼續進行比較。
<span style="font-size:18px;">#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</span><4> 測試
最後我們來測試一下:
資料量：short.MaxValue。
記憶體存放量：1200。
在這種場景下，我們決定每個檔案放1000條，也就有33個小檔案，也就有33個記憶體佇列，每個佇列取Top100條，Batch=500時重新整理
硬碟，中轉站存放33*2個數字（因為入中轉站時打上了佇列標記）,最後記憶體活動最大總數為：sum=33*100+500+66=896<1200。
時間複雜度為N*logN。當然這個“閥值”，我們可以再仔細微調。
<span style="font-size:18px;">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();
}</span>
總的程式碼：
<span style="font-size:18px;">using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Diagnostics;
using System.Threading;
using System.IO;
using System.Threading.Tasks;
namespace ConsoleApplication2
{
public class Program
{
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();
}
#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
#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
#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
}
}</span>優先佇列
<span style="font-size:18px;">using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Diagnostics;
using System.Threading;
using System.IO;
namespace ConsoleApplication2
{
public class PriorityQueue<T>
{
/// <summary>
/// 定義一個數組來存放節點
/// </summary>
private List<HeapNode> nodeList = new List<HeapNode>();
#region 堆節點定義
/// <summary>
/// 堆節點定義
/// </summary>
public class HeapNode
{
/// <summary>
/// 實體資料
/// </summary>
public T t { get; set; }
/// <summary>
/// 優先級別 1-10個級別 (優先級別遞增)
/// </summary>
public int level { get; set; }
public HeapNode(T t, int level)
{
this.t = t;
this.level = level;
}
public HeapNode() { }
}
#endregion
#region 新增操作
/// <summary>
/// 新增操作
/// </summary>
public void Eequeue(T t, int level = 1)
{
//將當前節點追加到堆尾
nodeList.Add(new HeapNode(t, level));
//如果只有一個節點，則不需要進行篩操作
if (nodeList.Count == 1)
return;
//獲取最後一個非葉子節點
int parent = nodeList.Count / 2 - 1;
//堆調整
UpHeapAdjust(nodeList, parent);
}
#endregion
#region 對堆進行上濾操作，使得滿足堆性質
/// <summary>
/// 對堆進行上濾操作，使得滿足堆性質
/// </summary>
/// <param name="nodeList"></param>
/// <param name="index">非葉子節點的之後指標（這裡要注意：我們
/// 的篩操作時針對非葉節點的）
/// </param>
public void UpHeapAdjust(List<HeapNode> nodeList, int parent)
{
while (parent >= 0)
{
//當前index節點的左孩子
var left = 2 * parent + 1;
//當前index節點的右孩子
var right = left + 1;
//parent子節點中最大的孩子節點，方便於parent進行比較
//預設為left節點
var min = left;
//判斷當前節點是否有右孩子
if (right < nodeList.Count)
{
//判斷parent要比較的最大子節點
min = nodeList[left].level < nodeList[right].level ? left : right;
}
//如果parent節點大於它的某個子節點的話，此時篩操作
if (nodeList[parent].level > nodeList[min].level)
{
//子節點和父節點進行交換操作
var temp = nodeList[parent];
nodeList[parent] = nodeList[min];
nodeList[min] = temp;
//繼續進行更上一層的過濾
parent = (int)Math.Ceiling(parent / 2d) - 1;
}
else
{
break;
}
}
}
#endregion
#region 優先佇列的出隊操作
/// <summary>
/// 優先佇列的出隊操作
/// </summary>
/// <returns></returns>
public HeapNode Dequeue()
{
if (nodeList.Count == 0)
return null;
//出佇列操作，彈出資料頭元素
var pop = nodeList[0];
//用尾元素填充頭元素
nodeList[0] = nodeList[nodeList.Count - 1];
//刪除尾節點
nodeList.RemoveAt(nodeList.Count - 1);
//然後從根節點下濾堆
DownHeapAdjust(nodeList, 0);
return pop;
}
#endregion
#region 對堆進行下濾操作，使得滿足堆性質
/// <summary>
/// 對堆進行下濾操作，使得滿足堆性質
/// </summary>
/// <param name="nodeList"></param>
/// <param name="index">非葉子節點的之後指標（這裡要注意：我們
/// 的篩操作時針對非葉節點的）
/// </param>
public void DownHeapAdjust(List<HeapNode> nodeList, int parent)
{
while (2 * parent + 1 < nodeList.Count)
{
//當前index節點的左孩子
var left = 2 * parent + 1;
//當前index節點的右孩子
var right = left + 1;
//parent子節點中最大的孩子節點，方便於parent進行比較
//預設為left節點
var min = left;
//判斷當前節點是否有右孩子
if (right < nodeList.Count)
{
//判斷parent要比較的最大子節點
min = nodeList[left].level < nodeList[right].level ? left : right;
}
//如果parent節點小於它的某個子節點的話，此時篩操作
if (nodeList[parent].level > nodeList[min].level)
{
//子節點和父節點進行交換操作
var temp = nodeList[parent];
nodeList[parent] = nodeList[min];
nodeList[min] = temp;
//繼續進行更下一層的過濾
parent = min;
}
else
{
break;
}
}
}
#endregion
#region 獲取元素並下降到指定的level級別
/// <summary>
/// 獲取元素並下降到指定的level級別
/// </summary>
/// <returns></returns>
public HeapNode GetAndDownPriority(int level)
{
if (nodeList.Count == 0)
return null;
//獲取頭元素
var pop = nodeList[0];
//設定指定優先順序（如果為 MinValue 則為 -- 操作）
nodeList[0].level = level == int.MinValue ? --nodeList[0].level : level;
//下濾堆
DownHeapAdjust(nodeList, 0);
return nodeList[0];
}
#endregion
#region 獲取元素並下降優先順序
/// <summary>
/// 獲取元素並下降優先順序
/// </summary>
/// <returns></returns>
public HeapNode GetAndDownPriority()
{
//下降一個優先順序
return GetAndDownPriority(int.MinValue);
}
#endregion
#region 返回當前優先佇列中的元素個數
/// <summary>
/// 返回當前優先佇列中的元素個數
/// </summary>
/// <returns></returns>
public int Count()
{
return nodeList.Count;
}
#endregion
}
}</span>