背景:

我正在清理制表符分隔的大文件(无法保存在内存中)。清理输入文件时,我会在内存中建立一个列表。当它达到1,000,000个条目(大约1GB的内存)时,我对其进行排序(使用下面的默认键)并将列表写入文件。此类用于将排序后的文件放回原处。它可以处理到目前为止我遇到的文件。到目前为止,我最大的案例是合并66个排序的文件。

问题:

  • 我的逻辑上是否有漏洞(哪里很脆弱)?
  • 我是否实现了合并排序
    算法正确吗?
  • 有什么明显的改进
    那可以做到吗?

  • 示例数据:

    这是这些文件之一中的一行的抽象:
    'hash_of_SomeStringId\tSome String Id\t\t\twww.somelink.com\t\tOtherData\t\n'
    要点是我使用'SomeStringId'.lower().replace(' ', '')作为我的排序键。

    原始代码:
    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
    

    编辑:实现Brian的建议后,我想到了以下解决方案:

    第二次编辑:更新了根据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
    

    粗测

    使用相同的输入文件(2.2 GB数据):
  • SortedFileMerger类花了51
    分钟(3068.4秒)
  • Brian的解决方案花费了40分钟(2408.5秒)
  • 添加John Machin的建议后,
    解决方案代码耗时36分钟
    (2214.0秒)
  • 最佳答案

    请注意,在python2.6中,heapq有一个新的merge函数,它将为您完成此操作。

    要处理自定义键功能,您只需将文件迭代器包装上装饰它的东西,以便它根据键进行比较,然后将其剥离:

    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])
    

    [编辑] 即使在较早版本的python中,简单地从更高版本的heapq模块中实现合并也是可能值得的。它是纯python,在python2.5中未经修改即可运行,并且由于它使用堆来获取下一个最小值,因此在合并大量文件时应该非常有效。

    您应该能够简单地从python2.6安装中复制heapq.py,将其作为“heapq26.py”复制到源代码中,并使用“from heapq26 import merge”-其中没有使用2.6特定功能。另外,您也可以复制合并功能(重写heappop等调用以引用python2.5 heapq模块)。

    关于合并分类文件的Python类,该如何改进?,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/1001569/

    10-11 22:39
    查看更多