Algoritmo para la combinación de N vías

Una combinación bidireccional se estudia ampliamente como parte del algoritmo Mergesort. Pero estoy interesado en averiguar cuál es la mejor manera de realizar una fusión de N vías.

Digamos que tengo Narchivos que han ordenado 1 millón de enteros cada uno. Tengo que fusionarlos en un solo archivo que tendrá esos 100 millones de enteros ordenados.

Tenga en cuenta que el caso de uso de este problema es en realidad la clasificación externa que se basa en disco. Por lo tanto, en escenarios reales también habría limitación de memoria. Por lo tanto, un enfoque ingenuo de fusionar 2 archivos a la vez (99 veces) no funcionará. Digamos que solo tenemos una pequeña ventana deslizante de memoria disponible para cada arreglo.

No estoy seguro de si ya existe una solución estandarizada para esta fusión de N vías. (Buscar en Google no me dijo mucho) .

Pero si sabe si es un buen algoritmo de fusión de n vías, publique algo / enlace.

Complejidad del tiempo: si aumentamos en gran medida la cantidad de archivos ( N) que se fusionarán, ¿cómo afectaría eso a la complejidad del tiempo de su algoritmo?

Gracias por tus respuestas.

No me han preguntado esto en ninguna parte, pero sentí que podría ser una pregunta interesante para la entrevista. Por lo tanto etiquetado.

¿Qué tal la siguiente idea?

1. Crea una cola de prioridad
   
   
2. Iterar a través de cada archivo f
   1. poner en cola el par (nextNumberIn (f), f) usando el primer valor como clave de prioridad
      
      
3. Mientras la cola no esté vacía
   1. dequeue head (m, f) de la cola
   2. salida m
   3. si f no agotado
      1. poner en cola (nextNumberIn (f), f)

Dado que la adición de elementos a una cola de prioridad se puede hacer en tiempo logarítmico, el elemento 2 es O (N × log N) . Dado que (casi todas) las iteraciones del ciclo while agregan un elemento, todo el ciclo while es O (M × log N) donde M es el número total de números para ordenar.

Suponiendo que todos los archivos tienen una secuencia de números no vacía, tenemos M> N y, por lo tanto, todo el algoritmo debería ser O (M × log N) .

Busque "Polyphase merge", vea los clásicos: Donald Knuth y EHFriend.

Además, es posible que desee echar un vistazo a la fusión de bloques inteligentes propuesta por Seyedafsari & Hasanzadeh, que, de manera similar a las sugerencias anteriores, usa colas de prioridad.

Otro razonamiento interesante es el algoritmo de fusión in situ de Kim & Kutzner.

También recomiendo este artículo de Vitter: Algoritmos de memoria externa y estructuras de datos: manejo de datos masivos .

Una idea simple es mantener una cola de prioridad de los rangos para fusionar, almacenada de tal manera que el rango con el primer elemento más pequeño se elimine primero de la cola. Luego puede hacer una combinación de N vías de la siguiente manera:

1. Inserte todos los rangos en la cola de prioridad, excepto los rangos vacíos.
2. Si bien la cola de prioridad no está vacía:
   1. Retirar de la cola el elemento más pequeño de la cola.
   2. Agregue el primer elemento de este rango a la secuencia de salida.
   3. Si no está vacío, inserte el resto de la secuencia nuevamente en la cola de prioridad.

La exactitud de este algoritmo es esencialmente una generalización de la prueba de que una combinación bidireccional funciona correctamente: si siempre agrega el elemento más pequeño de cualquier rango y todos los rangos están ordenados, terminará con la secuencia como un todo ordenado.

La complejidad del tiempo de ejecución de este algoritmo se puede encontrar de la siguiente manera. Sea M el número total de elementos en todas las secuencias. Si usamos un montón binario, entonces hacemos como máximo inserciones O (M) y eliminaciones O (M) de la cola de prioridad, ya que para cada elemento escrito en la secuencia de salida hay una salida de cola para extraer la secuencia más pequeña, seguida de poner en cola para poner el resto de la secuencia de nuevo en la cola. Cada uno de estos pasos requiere O (lg N) operaciones, porque la inserción o eliminación de un montón binario con N elementos lleva O (lg N) tiempo. Esto da un tiempo de ejecución neto de O (M lg N), que crece menos que linealmente con el número de secuencias de entrada.

Puede haber una forma de conseguirlo aún más rápido, pero parece una solución bastante buena. El uso de memoria es O (N) porque necesitamos O (N) sobrecarga para el montón binario. Si implementamos el montón binario almacenando punteros a las secuencias en lugar de a las secuencias en sí, esto no debería ser un gran problema a menos que tenga una cantidad realmente ridícula de secuencias para fusionar. En ese caso, simplemente combínelos en grupos que encajen en la memoria, luego combine todos los resultados.

¡Espero que esto ayude!

Un enfoque simple para fusionar k matrices ordenadas (cada una de longitud n) requiere O (nk ^ 2) tiempo y no O (nk) tiempo. Como cuando fusiona las primeras 2 matrices, lleva 2n tiempo, luego, cuando fusiona la tercera con la salida, toma 3n tiempo, ya que ahora estamos fusionando dos matrices de longitud 2n y n. Ahora, cuando fusionamos esta salida con la cuarta, esta fusión requiere 4n tiempo. Por lo tanto, la última fusión (cuando estamos agregando la matriz kth a nuestra matriz ya ordenada) requiere k * n tiempo. Por lo tanto, el tiempo total requerido es 2n + 3n + 4n + ... k * n que es O (nk ^ 2).

Parece que podemos hacerlo en tiempo O (kn) pero no es así porque cada vez que nuestro arreglo que estamos fusionando aumenta de tamaño.
Aunque podemos lograr un mejor límite usando divide y conquistar. Todavía estoy trabajando en eso y publico una solución si encuentro una.

Consulte http://en.wikipedia.org/wiki/External_sorting . Aquí está mi opinión sobre la combinación de k-way basada en el montón, usando una lectura almacenada en búfer de las fuentes para emular la reducción de E / S:

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

Aquí hay una posible implementación deMinHeap<T> . Algunas pruebas:

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

Escribí este código al estilo STL que se fusiona en N y pensé en publicarlo aquí para ayudar a evitar que otros reinventen la rueda. :)

Advertencia: solo se ha probado levemente. Prueba antes de usar. :)

Puedes usarlo así:

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

También permite usar pares de iteradores en lugar de los propios contenedores.

Si usa Boost.Range, puede eliminar parte del código repetitivo.

El código:

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

}

}

Implementación Java del algoritmo min heap para fusionar k matrices ordenadas:

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

}