在大型文本文件中搜索字符串的廉价方法

Question

我需要在一个相当大的文本文件中搜索特定的字符串。它是一个包含大约 5000 行文本的构建日志。这样做的最佳方法是什么？使用正则表达式应该不会引起任何问题吗？我将继续阅读行块，并使用简单的查找。

Answer 1

如果是“相当大”的文件，则按顺序访问这些行，不要将整个文件读入内存：

with open('largeFile', 'r') as inF:
    for line in inF:
        if 'myString' in line:
            # do_something

Answer 2

你可以做一个简单的查找：

f = open('file.txt', 'r')
lines = f.read()
answer = lines.find('string')

如果你能侥幸成功，一个简单的查找将比正则表达式快很多。

Answer 3

以下函数适用于文本文件和二进制文件（虽然只返回字节数中的位置），它确实有利于查找字符串，即使它们与行或缓冲区重叠并且不会在逐行或逐缓冲区搜索时找到。

def fnd(fname, s, start=0):
    with open(fname, 'rb') as f:
        fsize = os.path.getsize(fname)
        bsize = 4096
        buffer = None
        if start > 0:
            f.seek(start)
        overlap = len(s) - 1
        while True:
            if (f.tell() >= overlap and f.tell() < fsize):
                f.seek(f.tell() - overlap)
            buffer = f.read(bsize)
            if buffer:
                pos = buffer.find(s)
                if pos >= 0:
                    return f.tell() - (len(buffer) - pos)
            else:
                return -1

这背后的想法是：

• 在文件中寻找起始位置
• 从文件读取到缓冲区（搜索字符串必须小于缓冲区大小）但如果不是在开头，则回退 - 1 个字节，如果在最后一个读取缓冲区的末尾开始捕获字符串，并且继续下一个。
• 返回位置，如果没有找到则返回 -1

我使用类似这样的方法在较大的 ISO9660 文件中查找文件的签名，这非常快且不占用太多内存，您也可以使用更大的缓冲区来加快速度。

Answer 4

我尝试了一个文件文本搜索的多处理示例。这是我第一次尝试使用多处理模块；我是蟒蛇n00b。非常欢迎评论。我必须等到上班才能测试真正的大文件。它在多核系统上应该比单核搜索更快。呸！找到文本并可靠地报告行号后，如何停止进程？

import multiprocessing, os, time
NUMBER_OF_PROCESSES = multiprocessing.cpu_count()

def FindText( host, file_name, text):
    file_size = os.stat(file_name ).st_size 
    m1 = open(file_name, "r")

    #work out file size to divide up to farm out line counting

    chunk = (file_size / NUMBER_OF_PROCESSES ) + 1
    lines = 0
    line_found_at = -1

    seekStart = chunk * (host)
    seekEnd = chunk * (host+1)
    if seekEnd > file_size:
        seekEnd = file_size

    if host > 0:
        m1.seek( seekStart )
        m1.readline()

    line = m1.readline()

    while len(line) > 0:
        lines += 1
        if text in line:
            #found the line
            line_found_at = lines
            break
        if m1.tell() > seekEnd or len(line) == 0:
            break
        line = m1.readline()
    m1.close()
    return host,lines,line_found_at

# Function run by worker processes
def worker(input, output):
    for host,file_name,text in iter(input.get, 'STOP'):
        output.put(FindText( host,file_name,text ))

def main(file_name,text):
    t_start = time.time()
    # Create queues
    task_queue = multiprocessing.Queue()
    done_queue = multiprocessing.Queue()
    #submit file to open and text to find
    print 'Starting', NUMBER_OF_PROCESSES, 'searching workers'
    for h in range( NUMBER_OF_PROCESSES ):
        t = (h,file_name,text)
        task_queue.put(t)

    #Start worker processes
    for _i in range(NUMBER_OF_PROCESSES):
        multiprocessing.Process(target=worker, args=(task_queue, done_queue)).start()

    # Get and print results

    results = {}
    for _i in range(NUMBER_OF_PROCESSES):
        host,lines,line_found = done_queue.get()
        results[host] = (lines,line_found)

    # Tell child processes to stop
    for _i in range(NUMBER_OF_PROCESSES):
        task_queue.put('STOP')
#        print "Stopping Process #%s" % i

    total_lines = 0
    for h in range(NUMBER_OF_PROCESSES):
        if results[h][1] > -1:
            print text, 'Found at', total_lines + results[h][1], 'in', time.time() - t_start, 'seconds'
            break
        total_lines += results[h][0]

if __name__ == "__main__":
    main( file_name = 'testFile.txt', text = 'IPI1520' )

Answer 5

我很惊讶没有人提到将文件映射到内存：mmap

有了这个，您可以访问该文件，就好像它已经加载到内存中一样，并且操作系统会尽可能地将其映射进出。此外，如果您从 2 个独立进程执行此操作并且它们映射文件“共享”，它们将共享底层内存。

一旦映射，它将表现得像bytearray。您可以使用正则表达式、find 或任何其他常用方法。

请注意，这种方法有点特定于操作系统。它不会自动移植。

Answer 6

如果无法判断字符串的位置（前半部分、后半部分等），那么除了内置的“查找”功能之外，确实没有优化的方法来进行搜索。您可以通过不一次性读取文件，而是以 4kb 块（通常是硬盘块的大小）来减少 I/O 时间和内存消耗。这不会使搜索更快，除非字符串位于文件的第一部分，但无论如何都会减少内存消耗，如果文件很大，这可能是个好主意。

Answer 7

我喜欢 Javier 的解决方案。我没试过，但听起来很酷！

要阅读任意大文本并想知道它是否存在字符串，请将其替换，您可以使用Flashtext，这比使用非常大文件的正则表达式要快。

编辑：

来自开发者页面：

>>> from flashtext import KeywordProcessor
>>> keyword_processor = KeywordProcessor()
>>> # keyword_processor.add_keyword(<unclean name>, <standardised name>)
>>> keyword_processor.add_keyword('Big Apple', 'New York')
>>> keyword_processor.add_keyword('Bay Area')
>>> keywords_found = keyword_processor.extract_keywords('I love Big Apple and Bay Area.')
>>> keywords_found
>>> # ['New York', 'Bay Area']

或者提取偏移量的时候：

>>> from flashtext import KeywordProcessor
>>> keyword_processor = KeywordProcessor()
>>> keyword_processor.add_keyword('Big Apple', 'New York')
>>> keyword_processor.add_keyword('Bay Area')
>>> keywords_found = keyword_processor.extract_keywords('I love big Apple and Bay Area.', span_info=True)
>>> keywords_found
>>> # [('New York', 7, 16), ('Bay Area', 21, 29)]

Answer 8

这完全是受到laurasia's answer above 的启发，但它改进了结构。

它还添加了一些检查：

• 在空文件中搜索空字符串时会正确返回0。在 laurasia 的回答中，这是一个极端情况，将返回 -1。
• 它还会预先检查目标字符串是否大于缓冲区大小，如果是，则会引发错误。

在实践中，为了效率，目标字符串应该比缓冲区小得多，如果目标字符串的大小非常接近缓冲区的大小，还有更有效的搜索方法。

def fnd(fname, goal, start=0, bsize=4096):
    if bsize < len(goal):
        raise ValueError("The buffer size must be larger than the string being searched for.")
    with open(fname, 'rb') as f:
        if start > 0:
            f.seek(start)
        overlap = len(goal) - 1
        while True:
            buffer = f.read(bsize)
            pos = buffer.find(goal)
            if pos >= 0:
                return f.tell() - len(buffer) + pos
            if not buffer:
                return -1
            f.seek(f.tell() - overlap)

Answer 9

5000 行并不大（嗯，取决于行有多长...）

无论如何：假设字符串将是一个单词并且将由空格分隔...

lines=open(file_path,'r').readlines()
str_wanted="whatever_youre_looking_for"


    for i in range(len(lines)):
        l1=lines.split()
        for p in range(len(l1)):
            if l1[p]==str_wanted:
                #found
                # i is the file line, lines[i] is the full line, etc.