Algorytm scalania N-way

Łączenie dwukierunkowe jest szeroko badane jako część algorytmu Mergesort. Ale jestem zainteresowany, aby dowiedzieć się, w jaki sposób najlepiej wykonać scalanie N-way?

Powiedzmy, że mam Npliki, w których każdy posortował 1 milion liczb całkowitych. Muszę je scalić w jeden pojedynczy plik, który będzie zawierał te 100 milionów posortowanych liczb całkowitych.

Należy pamiętać, że przypadkiem użycia tego problemu jest w rzeczywistości sortowanie zewnętrzne oparte na dysku. Dlatego w rzeczywistych scenariuszach również wystąpiłoby ograniczenie pamięci. Tak więc naiwne podejście polegające na łączeniu 2 plików naraz (99 razy) nie zadziała. Powiedzmy, że dla każdej tablicy mamy tylko małe przesuwne okno pamięci.

Nie jestem pewien, czy istnieje już ustandaryzowane rozwiązanie tego połączenia N-way. (Googlowanie niewiele mi mówiło) .

Ale jeśli wiesz, czy dobry algorytm scalania n-way, prześlij algo / link.

Złożoność czasowa: jeśli znacznie zwiększymy liczbę plików ( N) do scalenia, jak wpłynie to na złożoność czasową algorytmu?

Dziękuję za odpowiedzi.

Nigdzie nie zadano mi tego, ale czułem, że może to być interesujące pytanie do wywiadu. Dlatego oznaczone.

A co z następującym pomysłem:

1. Utwórz kolejkę priorytetową
   
   
2. Przejdź przez każdy plik f
   1. umieść parę w kolejce (nextNumberIn (f), f) używając pierwszej wartości jako klucza priorytetu
      
      
3. Chociaż kolejka nie jest pusta
   1. dequeue head (m, f) of queue
   2. wyjście m
   3. jeśli f nie jest wyczerpany
      1. enqueue (nextNumberIn (f), f)

Ponieważ dodawanie elementów do kolejki priorytetowej może odbywać się w czasie logarytmicznym, pozycja 2 to O (N × log N) . Ponieważ (prawie wszystkie) iteracje pętli while dodają element, cała pętla while to O (M × log N), gdzie M to całkowita liczba liczb do sortowania.

Zakładając, że wszystkie pliki mają niepusty ciąg liczb, mamy M> N, a zatem cały algorytm powinien mieć wartość O (M × log N) .

Wyszukaj „Polyphase merge”, sprawdź klasykę - Donald Knuth & EHFriend.

Warto też przyjrzeć się propozycji Smart Block Merging autorstwa Seyedafsari i Hasanzadeh, która podobnie jak wcześniejsze sugestie wykorzystuje kolejki priorytetowe.

Innym interesującym powodem jest algorytm łączenia na miejscu autorstwa Kim & Kutzner.

Polecam również artykuł Vittera: Algorytmy pamięci zewnętrznej i struktury danych: radzenie sobie z ogromnymi danymi .

Jednym prostym pomysłem jest zachowanie kolejki priorytetowej zakresów do scalenia, przechowywanej w taki sposób, aby zakres z najmniejszym pierwszym elementem był usuwany jako pierwszy z kolejki. Następnie możesz wykonać scalanie N-way w następujący sposób:

1. Wstaw wszystkie zakresy do kolejki priorytetów, z wyłączeniem pustych zakresów.
2. Podczas gdy kolejka priorytetowa nie jest pusta:
   1. Usuń z kolejki najmniejszy element z kolejki.
   2. Dołącz pierwszy element tego zakresu do sekwencji wyjściowej.
   3. Jeśli jest niepusty, wstaw resztę sekwencji z powrotem do kolejki priorytetowej.

Poprawność tego algorytmu jest zasadniczo uogólnieniem dowodu na to, że scalanie dwukierunkowe działa poprawnie - jeśli zawsze dodajesz najmniejszy element z dowolnego zakresu, a wszystkie zakresy są posortowane, kończy się posortowaniem sekwencji jako całości.

Złożoność środowiska wykonawczego tego algorytmu można znaleźć w następujący sposób. Niech M będzie całkowitą liczbą elementów we wszystkich sekwencjach. Jeśli używamy stosu binarnego, to robimy co najwyżej O (M) wstawień i O (M) usunięć z kolejki priorytetowej, ponieważ dla każdego elementu zapisanego w sekwencji wyjściowej jest kolejka do wyciągnięcia najmniejszej sekwencji, po której następuje enqueue, aby umieścić resztę sekwencji z powrotem w kolejce. Każdy z tych kroków wymaga operacji O (lg N), ponieważ wstawienie lub usunięcie ze stosu binarnego zawierającego N elementów zajmuje O (lg N) czasu. Daje to czas wykonania netto O (M lg N), który rośnie mniej niż liniowo wraz z liczbą sekwencji wejściowych.

Być może istnieje sposób, aby to osiągnąć jeszcze szybciej, ale wydaje się to całkiem dobrym rozwiązaniem. Użycie pamięci wynosi O (N), ponieważ potrzebujemy O (N) narzutu dla sterty binarnej. Jeśli zaimplementujemy stertę binarną, przechowując wskaźniki do sekwencji, a nie do samych sekwencji, nie powinno to stanowić większego problemu, chyba że masz naprawdę śmieszną liczbę sekwencji do scalenia. W takim przypadku po prostu połącz je w grupy, które mieszczą się w pamięci, a następnie połącz wszystkie wyniki.

Mam nadzieję że to pomoże!

Proste podejście do scalania k posortowanych tablic (każda o długości n) wymaga O (nk ^ 2) czasu, a nie O (nk) czasu. Ponieważ łączenie pierwszych 2 tablic zajmuje 2n czasu, a kiedy łączysz trzecią tablicę z wynikiem, zajmuje to 3n czasu, ponieważ teraz łączymy dwie tablice o długości 2n i n. Teraz, kiedy połączymy to wyjście z czwartym, to scalenie wymaga 4n czasu, tak więc ostatnie scalenie (kiedy dodajemy k-tą tablicę do naszej już posortowanej tablicy) wymaga k * n czasu, a więc całkowity wymagany czas wynosi 2n + 3n + 4n + ... k * n, czyli O (nk ^ 2).

Wygląda na to, że możemy to zrobić w czasie O (kn), ale tak nie jest, ponieważ za każdym razem nasza tablica, którą scalamy, rośnie.
Chociaż możemy osiągnąć lepszą granicę za pomocą dziel i rządź. Nadal nad tym pracuję i publikuję rozwiązanie, jeśli je znajdę.

Zobacz http://en.wikipedia.org/wiki/External_sorting . Oto moje podejście do scalania k-way opartego na stercie, używającego buforowanego odczytu ze źródeł do emulacji redukcji we / wy:

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

Oto jedna możliwa implementacjaMinHeap<T> . Niektóre testy:

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

Napisałem ten fragment kodu w stylu STL, który łączy N-way scalanie i pomyślałem, że opublikuję go tutaj, aby pomóc innym uniemożliwić innym wymyślanie koła. :)

Ostrzeżenie: jest tylko lekko przetestowany. Przetestuj przed użyciem. :)

Możesz go używać w ten sposób:

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

Pozwala także na używanie par iteratorów zamiast samych kontenerów.

Jeśli używasz Boost.Range, możesz usunąć część kodu standardowego.

Kod:

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

}

}

Implementacja w Javie algorytmu minimalnej sterty do łączenia k posortowanych tablic:

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

}