Classe Python per unire i file ordinati, come può essere migliorato?

Question

Sfondo:

Sto pulendo file delimitati da tabulazioni grandi (impossibile da conservare in memoria). Mentre pulisco il file di input, creo un elenco in memoria; quando arriva a 1.000.000 di voci (circa 1 GB di memoria) l'ho ordinato (usando la chiave predefinita di seguito) e scrivo l'elenco in un file. Questa classe serve per rimettere insieme i file ordinati. Funziona sui file che ho incontrato finora. Il mio caso più grande, finora, è la fusione di 66 file ordinati.

Domande:

1. Ci sono buchi nella mia logica (dov'è fragile)?
2. Ho implementato il merge-sort algoritmo correttamente?
3. Ci sono evidenti miglioramenti che potrebbe essere fatto?

Dati di esempio:

Questa è un'astrazione di una linea in uno di questi file:

'hash_of_SomeStringId\tSome String Id\t\t\twww.somelink.com\t\tOtherData\t\n'

L'asporto è che uso 'SomeStringId'.lower().replace(' ', '') come chiave di ordinamento.

Codice originale:

class SortedFileMerger():
    """ A one-time use object that merges any number of smaller sorted 
        files into one large sorted file.

        ARGS:
            paths - list of paths to sorted files
            output_path - string path to desired output file
            dedup - (boolean) remove lines with duplicate keys, default = True
            key - use to override sort key, default = "line.split('\t')[1].lower().replace(' ', '')"
                  will be prepended by "lambda line: ".  This should be the same 
                  key that was used to sort the files being merged!
    """
    def __init__(self, paths, output_path, dedup=True, key="line.split('\t')[1].lower().replace(' ', '')"):
        self.key = eval("lambda line: %s" % key)
        self.dedup = dedup
        self.handles = [open(path, 'r') for path in paths]
        # holds one line from each file
        self.lines = [file_handle.readline() for file_handle in self.handles]
        self.output_file = open(output_path, 'w')
        self.lines_written = 0
        self._mergeSortedFiles() #call the main method

    def __del__(self):
        """ Clean-up file handles.
        """
        for handle in self.handles:
            if not handle.closed:
                handle.close()
        if self.output_file and (not self.output_file.closed):
            self.output_file.close()

    def _mergeSortedFiles(self):
        """ Merge the small sorted files to 'self.output_file'. This can 
            and should only be called once.
            Called from __init__().
        """
        previous_comparable = ''
        min_line = self._getNextMin()
        while min_line:
            index = self.lines.index(min_line)
            comparable = self.key(min_line)
            if not self.dedup:                      
                #not removing duplicates
                self._writeLine(index)
            elif comparable != previous_comparable: 
                #removing duplicates and this isn't one
                self._writeLine(index)
            else:                                   
                #removing duplicates and this is one
                self._readNextLine(index)
            previous_comparable = comparable
            min_line = self._getNextMin()
        #finished merging
        self.output_file.close()

    def _getNextMin(self):
        """ Returns the next "smallest" line in sorted order.
            Returns None when there are no more values to get.
        """
        while '' in self.lines:
            index = self.lines.index('')
            if self._isLastLine(index):
                # file.readline() is returning '' because 
                # it has reached the end of a file.
                self._closeFile(index)
            else:
                # an empty line got mixed in
                self._readNextLine(index)
        if len(self.lines) == 0:
            return None
        return min(self.lines, key=self.key)

    def _writeLine(self, index):
        """ Write line to output file and update self.lines
        """
        self.output_file.write(self.lines[index])
        self.lines_written += 1
        self._readNextLine(index)

    def _readNextLine(self, index):
        """ Read the next line from handles[index] into lines[index]
        """
        self.lines[index] = self.handles[index].readline()

    def _closeFile(self, index):
        """ If there are no more lines to get in a file, it 
            needs to be closed and removed from 'self.handles'.
            It's entry in 'self.lines' also need to be removed.
        """
        handle = self.handles.pop(index)
        if not handle.closed:
            handle.close()
        # remove entry from self.lines to preserve order
        _ = self.lines.pop(index)

    def _isLastLine(self, index):
        """ Check that handles[index] is at the eof.
        """
        handle = self.handles[index]            
        if handle.tell() == os.path.getsize(handle.name):
            return True
        return False

Modifica: implementando i suggerimenti di Brian ho trovato la seguente soluzione:

Seconda modifica: aggiornato il codice per suggerimento di John Machin :

def decorated_file(f, key):
    """ Yields an easily sortable tuple. 
    """
    for line in f:
        yield (key(line), line)

def standard_keyfunc(line):
    """ The standard key function in my application.
    """
    return line.split('\t', 2)[1].replace(' ', '').lower()

def mergeSortedFiles(paths, output_path, dedup=True, keyfunc=standard_keyfunc):
    """ Does the same thing SortedFileMerger class does. 
    """
    files = map(open, paths) #open defaults to mode='r'
    output_file = open(output_path, 'w')
    lines_written = 0
    previous_comparable = ''
    for line in heapq26.merge(*[decorated_file(f, keyfunc) for f in files]):
        comparable = line[0]
        if previous_comparable != comparable:
            output_file.write(line[1])
            lines_written += 1
        previous_comparable = comparable
    return lines_written

Ruvido Test

Uso degli stessi file di input (2,2 GB di dati):

• La classe SortedFileMerger ha richiesto 51 minuti (3068,4 secondi)
• La soluzione di Brian ha impiegato 40 minuti (2408,5 secondi)
• Dopo aver aggiunto i suggerimenti di John Machin , il codice della soluzione ha richiesto 36 minuti (2214,0 secondi)

Answer 1

Nota che in python2.6, heapq ha un nuovo unisci funzione che lo farà per te.

Per gestire la funzione di tasto personalizzato, puoi semplicemente avvolgere l'iteratore di file con qualcosa che lo decora in modo che si confronti in base al tasto e lo spogli in seguito:

def decorated_file(f, key):
    for line in f: 
        yield (key(line), line)

filenames = ['file1.txt','file2.txt','file3.txt']
files = map(open, filenames)
outfile = open('merged.txt')

for line in heapq.merge(*[decorated_file(f, keyfunc) for f in files]):
    outfile.write(line[1])

[Modifica] Anche nelle versioni precedenti di Python, probabilmente vale la pena semplicemente prendere l'implementazione della fusione dal modulo heapq successivo. È puro Python e funziona senza modifiche in Python2.5, e poiché utilizza un heap per ottenere il minimo successivo dovrebbe essere molto efficiente quando si uniscono numerosi file.

Dovresti essere in grado di copiare semplicemente heapq.py da un'installazione python2.6, copiarlo sul tuo sorgente come " heapq26.py " e usa " from heapq26 import merge " - non ci sono 2.6 funzionalità specifiche utilizzate in esso. In alternativa, puoi semplicemente copiare la funzione di unione (riscrivendo le chiamate heappop ecc. Per fare riferimento al modulo heapq python2.5).

Answer 2

lt &; Lt &; Questo & Quot; rispondi & Quot; è un commento sul codice risultante del richiedente originale > >

Suggerimento: l'uso di eval () è ummmm e ciò che stai facendo limita il chiamante a usare lambda - l'estrazione della chiave può richiedere più di una riga, e in ogni caso non hai bisogno della stessa funzione per il preliminare passo di ordinamento?

Quindi sostituisci questo:

def mergeSortedFiles(paths, output_path, dedup=True, key="line.split('\t')[1].lower().replace(' ', '')"):
    keyfunc = eval("lambda line: %s" % key)

con questo:

def my_keyfunc(line):
    return line.split('\t', 2)[1].replace(' ', '').lower()
    # minor tweaks may speed it up a little

def mergeSortedFiles(paths, output_path, keyfunc, dedup=True):