Алгоритм N-стороннего слияния

Двустороннее слияние широко изучается как часть алгоритма сортировки слиянием. Но мне интересно узнать, как лучше всего выполнить N-образное слияние?

Допустим, у меня есть Nфайлы, в каждом из которых содержится 1 миллион целых чисел. Мне нужно объединить их в один файл, в котором будут эти 100 миллионов отсортированных целых чисел.

Имейте в виду, что вариант использования этой проблемы - это внешняя сортировка, основанная на диске. Следовательно, в реальных сценариях также будет ограничение памяти. Таким образом, наивный подход слияния двух файлов за раз (99 раз) не сработает. Допустим, у нас есть только небольшое скользящее окно памяти, доступное для каждого массива.

Я не уверен, есть ли уже стандартизированное решение для этого N-образного слияния. (Гугл мне мало что сказал) .

Но если вы знаете, есть ли хороший алгоритм n-way слияния, отправьте algo / link.

Сложность по времени: если мы значительно увеличим количество Nобъединяемых файлов ( ), как это повлияет на временную сложность вашего алгоритма?

Спасибо за ответы.

Меня нигде об этом не спрашивали, но я подумал, что это может быть интересный вопрос для интервью. Поэтому помечен.

Как насчет следующей идеи:

1. Создать приоритетную очередь
   
   
2. Перебрать каждый файл f
   1. поставить в очередь пару (nextNumberIn (f), f), используя первое значение в качестве ключа приоритета
      
      
3. Пока очередь не пуста
   1. исключить голову (m, f) очереди
   2. выход m
   3. если f не истощен
      1. enqueue (nextNumberIn (f), f)

Поскольку добавление элементов в приоритетную очередь может быть выполнено за логарифмическое время, элемент 2 равен O (N × log N) . Поскольку (почти все) итерации цикла while добавляют элемент, весь цикл while составляет O (M × log N), где M - общее количество чисел для сортировки.

Предполагая, что все файлы имеют непустую последовательность чисел, мы имеем M> N и, следовательно, весь алгоритм должен быть O (M × log N) .

Найдите "Полифазное слияние", посмотрите классику - Дональд Кнут и EHFriend.

Кроме того, вы можете взглянуть на предлагаемое слияние смарт-блоков, предложенное Сейедафсари и Хасанзаде, которое, как и предыдущие предложения, использует очереди приоритетов.

Еще один интересный аргумент - это алгоритм слияния на месте, разработанный Кимом и Кутцнером.

Я также рекомендую эту статью Виттера: Алгоритмы внешней памяти и структуры данных: работа с большими данными .

Одна простая идея - сохранить приоритетную очередь диапазонов для слияния, сохраненную таким образом, чтобы диапазон с наименьшим первым элементом удалялся первым из очереди. Затем вы можете выполнить N-стороннее слияние следующим образом:

1. Вставьте все диапазоны в приоритетную очередь, за исключением пустых диапазонов.
2. Пока очередь приоритетов не пуста:
   1. Удаляет из очереди самый маленький элемент.
   2. Добавьте первый элемент этого диапазона в выходную последовательность.
   3. Если он непустой, вставьте оставшуюся часть последовательности обратно в очередь приоритетов.

Правильность этого алгоритма по сути является обобщением доказательства того, что двухстороннее слияние работает правильно - если вы всегда добавляете наименьший элемент из любого диапазона, и все диапазоны отсортированы, вы в конечном итоге получаете отсортированную последовательность в целом.

Сложность выполнения этого алгоритма можно определить следующим образом. Пусть M - общее количество элементов во всех последовательностях. Если мы используем двоичную кучу, то мы делаем не более O (M) вставок и O (M) удалений из очереди с приоритетом, поскольку для каждого элемента, записанного в выходную последовательность, есть удаление из очереди для извлечения наименьшей последовательности, за которым следует enqueue, чтобы вернуть оставшуюся часть последовательности в очередь. Каждый из этих шагов требует O (lg N) операций, потому что вставка или удаление из двоичной кучи с N элементами занимает время O (lg N). Это дает чистое время выполнения O (M lg N), которое растет менее чем линейно с количеством входных последовательностей.

Возможно, есть способ сделать это еще быстрее, но это кажется довольно хорошим решением. Использование памяти составляет O (N), потому что нам нужны накладные расходы O (N) для двоичной кучи. Если мы реализуем двоичную кучу, сохраняя указатели на последовательности, а не на сами последовательности, это не должно быть большой проблемой, если у вас нет действительно смешного количества последовательностей для слияния. В этом случае просто объедините их в группы, которые умещаются в памяти, а затем объедините все результаты.

Надеюсь это поможет!

Простой подход к объединению k отсортированных массивов (каждый длиной n) требует времени O (nk ^ 2), а не времени O (nk). Когда вы объединяете первые 2 массива, это занимает 2n времени, тогда, когда вы объединяете третий с выходом, это занимает 3n времени, так как теперь мы объединяем два массива длиной 2n и n. Теперь, когда мы объединяем этот вывод с четвертым, это слияние требует 4n времени. Таким образом, последнее слияние (когда мы добавляем k-й массив в наш уже отсортированный массив) требует времени k * n. Таким образом, общее необходимое время составляет 2n + 3n + 4n + ... k * n, который равен O (nk ^ 2).

Похоже, мы можем сделать это за O (kn) раз, но это не так, потому что каждый раз наш объединяемый массив увеличивается в размере.
Хотя мы можем достичь лучшего результата, используя «разделяй и властвуй». Я все еще работаю над этим и опубликую решение, если найду его.

См. Http://en.wikipedia.org/wiki/External_sorting . Вот мой подход к слиянию k-way на основе кучи, с использованием буферизованного чтения из источников для имитации сокращения ввода-вывода:

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> . Некоторые тесты:

[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));
}

Я написал этот фрагмент кода в стиле STL, который выполняет N-образное слияние, и решил опубликовать его здесь, чтобы помочь другим не изобретать колесо. :)

Предупреждение: это только слегка протестировано. Перед использованием проверьте. :)

Вы можете использовать это так:

#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);
}

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);
    }

}

}

Реализация в Java алгоритма min heap для объединения k отсортированных массивов:

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);
}

}