N路合并algorithm

作为Mergesortalgorithm的一部分，双向合并被广泛​​研究。 但是我有兴趣找出一个可以进行N路合并的最佳方法吗？

可以说，我有N文件，每个文件已经sorting了100万个整数。 我必须将它们合并成1个单个文件，这将有1亿个sorting的整数。

请记住，这个问题的用例实际上是基于磁盘的外部sorting。 因此，在实际情况下也会有内存限制。 所以一次合并两个文件（99次）的一个天真的方法将不起作用。 可以说，我们只有一个可用于每个arrays的小滑动窗口。

我不确定是否已经有一个标准化的解决scheme来进行这种N路合并。 （谷歌search没有告诉我很多） 。

但是如果你知道一个好的n路合并algorithm，请发表algorithm/链接。

时间复杂度：如果我们大大增加要合并的文件数量（ N ），那将如何影响algorithm的时间复杂度？

感谢您的回答。

我从来没有被问到过这个问题，但是我觉得这可能是一个有趣的面试问题。 因此被标记。

Answer 1

下面的想法如何？

1. 创build一个优先队列
2. 迭代每个文件f
   1. 使用第一个值作为优先键入队（nextNumberIn（f），f）
3. 虽然队列不是空的
   1. 出队队列头（m，f）
   2. 输出m
   3. 如果f没有耗尽
      1. 入队（nextNumberIn（f），f）

由于将元素添加到优先队列可以在对数时间内完成，因此项目2是O（N×log N） 。 由于（几乎所有）while循环的迭代都添加了一个元素，整个while循环是O（M×logN） ，其中M是要sorting的总数。

假设所有文件都有一个非空的数字序列，我们有M> N ，因此整个algorithm应该是O（M×logN） 。

Answer 2

search“多相合并”，检查经典 – 唐纳德·克努特和EHFriend。

另外，您可能想看看Seyedafsari＆Hasanzadeh提出的Smart Block Merging ，与之前的build议类似，它使用优先级队列。

另一个有趣的原因是Kim＆Kutzner的In Place Merging Algorithm 。

我也推荐Vitter的这篇论文： 外部存储器algorithm和数据结构：处理海量数据 。

Answer 3

一个简单的想法是保持合并范围的优先级队列，以这样的方式存储，即首先从队列中移除具有最小的第一个元素的范围。 然后你可以做一个N路合并，如下所示：

1. 将所有范围插入优先级队列，不包括空白范围。
2. 虽然优先队列不是空的：
   1. 从队列中取出最小的元素。
   2. 将该范围的第一个元素附加到输出序列。
   3. 如果不是空的，则将序列的其余部分插回到优先级队列中。

这个algorithm的正确性本质上是一个双向合并工作正确的certificate的推广 – 如果你总是添加来自任何范围的最小元素，并且所有的范围都被sorting，那么最终将整个序列sorting。

该algorithm的运行时复杂度可以如下find。 设M是所有序列中元素的总数。 如果我们使用二进制堆，那么我们至多从优先级队列中进行O（M）插入和O（M）删除操作，因为对写入输出序列的每个元素都有一个出列队列来提取最小序列，接着是一个排队将序列的其余部分放回队列中。 每个步骤都需要O（lg N）操作，因为插入或删除一个二进制堆的N个元素需要花费O（lg N）的时间。 这产生了O（M lg N）的净运行时间，其与input序列的数量成线性地增长。

可能有办法让这个更快，但这似乎是一个很好的解决scheme。 内存使用量为O（N），因为我们需要二进制堆的O（N）开销。 如果我们通过存储指向序列的指针而不是序列本身来实现二进制堆，那么这应该不是一个太大的问题，除非你有一个真正荒谬的序列来合并。 在这种情况下，只需将它们合并到内存中，然后合并所有结果即可。

希望这可以帮助！

Answer 4

合并k个sorting数组（每个长度为n）的简单方法需要O（nk ^ 2）时间而不是O（nk）时间。 当你合并前两个数组需要2n时间，那么当你将第三个数据与输出合并时，需要3n时间，因为现在我们正在合并两个长度为2n和n的数组。 现在，当我们把这个输出与第四个合并时，这个合并需要4n时间。因此，最后的合并（当我们将第k个数组添加到已经sorting的数组时）需要k * n个时间。因此，所需的总时间是2n + 3n + 4n + … k * n是O（nk ^ 2）。

看起来我们可以在O（kn）时间做到这一点，但并不是这样，因为每次我们正在合并的数组的大小都在增加。
虽然我们可以通过分而治之来达到更好的界限。 我仍然在努力，并发布一个解决scheme，如果我find一个。

Answer 5

见http://en.wikipedia.org/wiki/External_sorting 。 这里是我的基于堆的k路合并，使用来自源的缓冲读取来模拟I / O减less：

 public class KWayMerger<T> { private readonly IList<T[]> _sources; private readonly int _bufferSize; private readonly MinHeap<MergeValue<T>> _mergeHeap; private readonly int[] _indices; public KWayMerger(IList<T[]> sources, int bufferSize, Comparer<T> comparer = null) { if (sources == null) throw new ArgumentNullException("sources"); _sources = sources; _bufferSize = bufferSize; _mergeHeap = new MinHeap<MergeValue<T>>( new MergeComparer<T>(comparer ?? Comparer<T>.Default)); _indices = new int[sources.Count]; } public T[] Merge() { for (int i = 0; i <= _sources.Count - 1; i++) AddToMergeHeap(i); var merged = new T[_sources.Sum(s => s.Length)]; int mergeIndex = 0; while (_mergeHeap.Count > 0) { var min = _mergeHeap.ExtractDominating(); merged[mergeIndex++] = min.Value; if (min.Source != -1) //the last item of the source was extracted AddToMergeHeap(min.Source); } return merged; } private void AddToMergeHeap(int sourceIndex) { var source = _sources[sourceIndex]; var start = _indices[sourceIndex]; var end = Math.Min(start + _bufferSize - 1, source.Length - 1); if (start > source.Length - 1) return; //we're done with this source for (int i = start; i <= end - 1; i++) _mergeHeap.Add(new MergeValue<T>(-1, source[i])); //only the last item should trigger the next buffered read _mergeHeap.Add(new MergeValue<T>(sourceIndex, source[end])); _indices[sourceIndex] += _bufferSize; //we may have added less items, //but if we did we've reached the end of the source so it doesn't matter } } internal class MergeValue<T> { public int Source { get; private set; } public T Value { get; private set; } public MergeValue(int source, T value) { Value = value; Source = source; } } internal class MergeComparer<T> : IComparer<MergeValue<T>> { public Comparer<T> Comparer { get; private set; } public MergeComparer(Comparer<T> comparer) { if (comparer == null) throw new ArgumentNullException("comparer"); Comparer = comparer; } public int Compare(MergeValue<T> x, MergeValue<T> y) { Debug.Assert(x != null && y != null); return Comparer.Compare(x.Value, y.Value); } } 

这里是MinHeap<T>一个可能的实现 。 一些testing：

 [TestMethod] public void TestKWaySort() { var rand = new Random(); for (int i = 0; i < 10; i++) AssertKwayMerge(rand); } private static void AssertKwayMerge(Random rand) { var sources = new[] { GenerateRandomCollection(rand, 10, 30, 0, 30).OrderBy(i => i).ToArray(), GenerateRandomCollection(rand, 10, 30, 0, 30).OrderBy(i => i).ToArray(), GenerateRandomCollection(rand, 10, 30, 0, 30).OrderBy(i => i).ToArray(), GenerateRandomCollection(rand, 10, 30, 0, 30).OrderBy(i => i).ToArray(), }; Assert.IsTrue(new KWayMerger<int>(sources, 20).Merge().SequenceEqual(sources.SelectMany(s => s).OrderBy(i => i))); } public static IEnumerable<int> GenerateRandomCollection(Random rand, int minLength, int maxLength, int min = 0, int max = int.MaxValue) { return Enumerable.Repeat(0, rand.Next(minLength, maxLength)).Select(i => rand.Next(min, max)); } 

Answer 6

我写了这个STL风格的N段合并代码，并且认为我会在这里发布它，以防止其他人重新发明轮子。 🙂

警告：这只是轻度testing。 使用前testing。 🙂

你可以像这样使用它：

 #include <vector> int main() { std::vector<std::vector<int> > v; std::vector<std::vector<int>::iterator> vout; std::vector<int> v1; std::vector<int> v2; v1.push_back(1); v1.push_back(2); v1.push_back(3); v2.push_back(0); v2.push_back(1); v2.push_back(2); v.push_back(v1); v.push_back(v2); multiway_merge(v.begin(), v.end(), std::back_inserter(vout), false); } 

它也允许使用迭代器对代替容器本身。

如果使用Boost.Range，则可以删除一些样板代码。

代码：

 #include <algorithm> #include <functional> // std::less #include <iterator> #include <queue> // std::priority_queue #include <utility> // std::pair #include <vector> template<class OutIt> struct multiway_merge_value_insert_iterator : public std::iterator< std::output_iterator_tag, OutIt, ptrdiff_t > { OutIt it; multiway_merge_value_insert_iterator(OutIt const it = OutIt()) : it(it) { } multiway_merge_value_insert_iterator &operator++(int) { return *this; } multiway_merge_value_insert_iterator &operator++() { return *this; } multiway_merge_value_insert_iterator &operator *() { return *this; } template<class It> multiway_merge_value_insert_iterator &operator =(It const i) { *this->it = *i; ++this->it; return *this; } }; template<class OutIt> multiway_merge_value_insert_iterator<OutIt> multiway_merge_value_inserter(OutIt const it) { return multiway_merge_value_insert_iterator<OutIt>(it); }; template<class Less> struct multiway_merge_value_less : private Less { multiway_merge_value_less(Less const &less) : Less(less) { } template<class It1, class It2> bool operator()( std::pair<It1, It1> const &b /* inverted */, std::pair<It2, It2> const &a) const { return b.first != b.second && ( a.first == a.second || this->Less::operator()(*a.first, *b.first)); } }; struct multiway_merge_default_less { template<class T> bool operator()(T const &a, T const &b) const { return std::less<T>()(a, b); } }; template<class R> struct multiway_merge_range_iterator { typedef typename R::iterator type; }; template<class R> struct multiway_merge_range_iterator<R const> { typedef typename R::const_iterator type; }; template<class It> struct multiway_merge_range_iterator<std::pair<It, It> > { typedef It type; }; template<class R> typename R::iterator multiway_merge_range_begin(R &r) { return r.begin(); } template<class R> typename R::iterator multiway_merge_range_end(R &r) { return r.end(); } template<class R> typename R::const_iterator multiway_merge_range_begin(R const &r) { return r.begin(); } template<class R> typename R::const_iterator multiway_merge_range_end(R const &r) { return r.end(); } template<class It> It multiway_merge_range_begin(std::pair<It, It> const &r) { return r.first; } template<class It> It multiway_merge_range_end(std::pair<It, It> const &r) { return r.second; } template<class It, class OutIt, class Less, class PQ> OutIt multiway_merge( It begin, It const end, OutIt out, Less const &less, PQ &pq, bool const distinct = false) { while (begin != end) { pq.push(typename PQ::value_type( multiway_merge_range_begin(*begin), multiway_merge_range_end(*begin))); ++begin; } while (!pq.empty()) { typename PQ::value_type top = pq.top(); pq.pop(); if (top.first != top.second) { while (!pq.empty() && pq.top().first == pq.top().second) { pq.pop(); } if (!distinct || pq.empty() || less(*pq.top().first, *top.first) || less(*top.first, *pq.top().first)) { *out = top.first; ++out; } ++top.first; pq.push(top); } } return out; } template<class It, class OutIt, class Less> OutIt multiway_merge( It const begin, It const end, OutIt out, Less const &less, bool const distinct = false) { typedef typename multiway_merge_range_iterator< typename std::iterator_traits<It>::value_type >::type SubIt; if (std::distance(begin, end) < 16) { typedef std::vector<std::pair<SubIt, SubIt> > Remaining; Remaining remaining; remaining.reserve( static_cast<size_t>(std::distance(begin, end))); for (It i = begin; i != end; ++i) { if (multiway_merge_range_begin(*i) != multiway_merge_range_end(*i)) { remaining.push_back(std::make_pair( multiway_merge_range_begin(*i), multiway_merge_range_end(*i))); } } while (!remaining.empty()) { typename Remaining::iterator smallest = remaining.begin(); for (typename Remaining::iterator i = remaining.begin(); i != remaining.end(); ) { if (less(*i->first, *smallest->first)) { smallest = i; ++i; } else if (distinct && i != smallest && !less( *smallest->first, *i->first)) { i = remaining.erase(i); } else { ++i; } } *out = smallest->first; ++out; ++smallest->first; if (smallest->first == smallest->second) { smallest = remaining.erase(smallest); } } return out; } else { std::priority_queue< std::pair<SubIt, SubIt>, std::vector<std::pair<SubIt, SubIt> >, multiway_merge_value_less<Less> > q((multiway_merge_value_less<Less>(less))); return multiway_merge(begin, end, out, less, q, distinct); } } template<class It, class OutIt> OutIt multiway_merge( It const begin, It const end, OutIt const out, bool const distinct = false) { return multiway_merge( begin, end, out, multiway_merge_default_less(), distinct); } 

Answer 7

 Here is my implementation using MinHeap... package merging; import java.io.BufferedReader; import java.io.BufferedWriter; import java.io.File; import java.io.FileReader; import java.io.FileWriter; import java.io.IOException; import java.io.PrintWriter; public class N_Way_Merge { int No_of_files=0; String[] listString; int[] listIndex; PrintWriter pw; private String fileDir = "D:\\XMLParsing_Files\\Extracted_Data"; private File[] fileList; private BufferedReader[] readers; public static void main(String[] args) throws IOException { N_Way_Merge nwm=new N_Way_Merge(); long start= System.currentTimeMillis(); try { nwm.createFileList(); nwm.createReaders(); nwm.createMinHeap(); } finally { nwm.pw.flush(); nwm.pw.close(); for (BufferedReader readers : nwm.readers) { readers.close(); } } long end = System.currentTimeMillis(); System.out.println("Files merged into a single file.\nTime taken: "+((end-start)/1000)+"secs"); } public void createFileList() throws IOException { //creates a list of sorted files present in a particular directory File folder = new File(fileDir); fileList = folder.listFiles(); No_of_files=fileList.length; assign(); System.out.println("No. of files - "+ No_of_files); } public void assign() throws IOException { listString = new String[No_of_files]; listIndex = new int[No_of_files]; pw = new PrintWriter(new BufferedWriter(new FileWriter("D:\\XMLParsing_Files\\Final.txt", true))); } public void createReaders() throws IOException { //creates array of BufferedReaders to read the files readers = new BufferedReader[No_of_files]; for(int i=0;i<No_of_files;++i) { readers[i]=new BufferedReader(new FileReader(fileList[i])); } } public void createMinHeap() throws IOException { for(int i=0;i<No_of_files;i++) { listString[i]=readers[i].readLine(); listIndex[i]=i; } WriteToFile(listString,listIndex); } public void WriteToFile(String[] listString,int[] listIndex) throws IOException{ BuildHeap_forFirstTime(listString, listIndex); while(!(listString[0].equals("zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz"))) { pw.println(listString[0]); listString[0]=readers[listIndex[0]].readLine(); MinHeapify(listString,listIndex,0); } } public void BuildHeap_forFirstTime(String[] listString,int[] listIndex){ for(int i=(No_of_files/2)-1;i>=0;--i) MinHeapify(listString,listIndex,i); } public void MinHeapify(String[] listString,int[] listIndex,int index){ int left=index*2 + 1; int right=left + 1; int smallest=index; int HeapSize=No_of_files; if(left <= HeapSize-1 && listString[left]!=null && (listString[left].compareTo(listString[index])) < 0) smallest = left; if(right <= HeapSize-1 && listString[right]!=null && (listString[right].compareTo(listString[smallest])) < 0) smallest=right; if(smallest!=index) { String temp=listString[index]; listString[index]=listString[smallest]; listString[smallest]=temp; listIndex[smallest]^=listIndex[index]; listIndex[index]^=listIndex[smallest]; listIndex[smallest]^=listIndex[index]; MinHeapify(listString,listIndex,smallest); } } 

}

Answer 8

用于合并k个sorting数组的min堆algorithm的Java实现：

 public class MergeKSorted { /** * helper object to store min value of each array in a priority queue, * the kth array and the index into kth array * */ static class PQNode implements Comparable<PQNode>{ int value; int kth = 0; int indexKth = 0; public PQNode(int value, int kth, int indexKth) { this.value = value; this.kth = kth; this.indexKth = indexKth; } @Override public int compareTo(PQNode o) { if(o != null) { return Integer.valueOf(value).compareTo(Integer.valueOf(o.value)); } else return 0; } @Override public String toString() { return value+" "+kth+" "+indexKth; } } public static void mergeKSorted(int[][] sortedArrays) { int k = sortedArrays.length; int resultCtr = 0; int totalSize = 0; PriorityQueue<PQNode> pq = new PriorityQueue<>(); for(int i=0; i<k; i++) { int[] kthArray = sortedArrays[i]; totalSize+=kthArray.length; if(kthArray.length > 0) { PQNode temp = new PQNode(kthArray[0], i, 0); pq.add(temp); } } int[] result = new int[totalSize]; while(!pq.isEmpty()) { PQNode temp = pq.poll(); int[] kthArray = sortedArrays[temp.kth]; result[resultCtr] = temp.value; resultCtr++; temp.indexKth++; if(temp.indexKth < kthArray.length) { temp = new PQNode(kthArray[temp.indexKth], temp.kth, temp.indexKth); pq.add(temp); } } print(result); } public static void print(int[] a) { StringBuilder sb = new StringBuilder(); for(int v : a) { sb.append(v).append(" "); } System.out.println(sb); } public static void main(String[] args) { int[][] sortedA = { {3,4,6,9}, {4,6,8,9,12}, {3,4,9}, {1,4,9} }; mergeKSorted(sortedA); } }