使用内置函数和并行化读取大型 (>5GB) 日志文件的最快方法？

Question

作为 python 新手，我的任务是找到在 Python 中解析大型日志文件的最快方法。

这些是我迄今为止尝试过的方法，它们给了我 33 到 43 秒的处理时间。 这个耗时最长，为 43 秒：

tflines1 = tfile.readlines()

time_data_count = 0
for line in tflines1:
    if 'TIME_DATA' in line :
        time_data_count += 1
if time_data_count > 20 :
    print("time_data complete")
else:
    print("incomplete time_data data")

这个平均耗时 34 秒：

with open(filename) as f:
    time_data_count = 0
    while True:
        memcap = f.read(102400)
        memcaplist = memcap.split("\n")
        for line in memcaplist:
            if 'TIME_DATA' in line:
                time_data_count += 1
        if not memcap:
            break

这个平均 36 秒：

with open(filename, 'r', buffering=102400) as f:
    time_data_count = 0
    for line in f:
        if 'TIME_DATA' in line:
            time_data_count += 1

这个平均 36 秒：

logfile = open(filename)
time_data_count = 0
for line in logfile:
    if 'TIME_DATA' in line:
            time_data_count += 1

最快的是这个在 26.8 秒内完成任务的，我不知道为什么它是最快的。我不明白是什么让它如此特别。有了这个和我指定字节的其他类似文件，我担心可能有一两个文件在字节块之间分割一行，而我正在寻找的字符串被分成两半。这将导致错误的计数。如何解决这个问题：

with open(filename) as f:
    time_data_count = 0
    while True:
        memcap = f.read(102400)
        time_data_count += memcap.count('TIME_DATA')
        if not memcap:
            break
    if time_data_count > 20:
        print("time_data complete")
    else:
        print("incomplete time_data data")

无论如何，老板告诉我要研究其他可能使事情变得更快的方法。他建议列表理解和以二进制形式读取文件对象。我认为列表理解不会有太大帮助，并且能够从文件中提取我需要的数据。我觉得将文件作为二进制文件读取会产生额外的代码行和代码需要知道何时预测某些字符的问题。甚至以二进制形式读取文件会有所不同吗？这不是可以使用指针的 c++。

我简要阅读了并行化，但我不确定这是否适用于我们的用例。我需要跟踪某些字符串出现了多少次，所以我不确定当你想跟踪事物的数量时如何将文件拆分为不同的线程。这甚至可能吗？

@TimPeters 这里是使用二进制文件和 seek() 对最终方法的编辑：

with open(filename, 'rb') as f:
    time_data_count = 0
    while True:

        memcap = f.read(102400)
        f.seek(-tdatlength, 1)
        time_data_count += memcap.count(b'TIME_DATA')

        if not memcap:
            break
    if time_data_count > 20:
        print("time_data complete")
    else:
        print("incomplete time_data data")

我尝试的另一种方法：

with open(filename, 'rb', buffering=102400) as f:
    time_data_count = 0
      #ask tenzin about seek in this situation
    for line in f:
        if b'TIME_DATA' in line:
            time_data_count += 1
    f.seek(-tdatlength, 2)
    if time_data_count > 20:
        print("time_data complete")
    else:
        print("incomplete time_data data")
    print(time_data_count)

Answer 1

您可以使用 python 的 multiprocessing 库并行解析多个日志文件：

import multiprocessing

def process_log(filename):
    with open(filename) as f:
        time_data_count = 0
        while True:
            memcap = f.read(102400)
            time_data_count += memcap.count('TIME_DATA')
            if not memcap:
                break
        if time_data_count > 20:
            print("time_data complete")
            return True
        else:
            print("incomplete time_data data")
            return False

filepaths = # load all the filepaths here with something like glob.glob("path/to/logdir/*.log)
pool = multiprocessing.Pool(num_cores_to_use) # set num cores to use

number_of_complete_logs = 0
for complete in pool.imap_unordered(process_log, filepaths):
    if complete:
        number_of_complete_logs += 1

print(number_of_complete_logs)

这样，如果您有 4 个内核，您将在 1 的时间内处理 4 个日志文件。另外，每个文件都是单独处理的，因此 TIME_DATA 计数器保持不变。

如果您要处理大量日志文件，我建议使用tqdm：

for complete in tqdm(pool.imap_unordered(process_log, filepaths), total=len(filepaths)):

这样您可以跟踪进度并估计整个操作需要多长时间。

Answer 2

这里使用mmap 来提高I/O 性能。如果没有您的日志文件，我无法对其进行基准测试，但我相信它会比基于行的 I/O 快得多。

如果你想并行化它，你可以扩展它，因为 mmap 对象支持随机访问（参见seek()）。因此，您可以启动多个线程同时从文件中的多个点开始搜索。

import mmap
import sys

with open(sys.argv[1], 'rb') as f:
    mm = mmap.mmap(f.fileno(), 0, prot=mmap.PROT_READ)
    target = b'TIME_DATA'
    tl = len(target)
    idx = -tl
    counter = 0
    while True:
        idx = mm.find(target, idx + tl)
        if idx < 0:
            break
        counter += 1
    print(counter)
    mm.close()

Answer 3

提示在这里不起作用，因此以具体的方式充实检查重叠的想法，相当于 - 但不使用 - .seek()。

target = b"TIME_DATA"
windowsize = len(target) - 1
last = b""
target_count = 0
with open(filename, "rb") as f:
    while True:
        memcap = f.read(102400)
        if not memcap:
            break
        overlap = last[-windowsize :] + memcap[: windowsize]
        if target in overlap:
            target_count += 1
        target_count += memcap.count(target)
        last = memcap

windowsize 对于target 来说太小了，无法匹配取自last 的overlap 部分或来自memcap 的部分，因此如果在target 中找到target，它必须在每个部分中至少匹配一个字符：它确实是重叠匹配。在另一个方向，如果相邻块之间存在匹配，它必须从last 的最后一个windowsize 字符之一开始，并在memcap 的第一个windowsize 字符之一结束，所以windowsize 是大到可以找到任何这样的匹配项。

编辑：修复了以下过于强烈的声明。

有一个不明确的地方：如果target 的某个前缀也是target 的后缀，则匹配可以重叠。例如，“abab”有“ab”作为前缀和后缀。因此，如果一个块以“abab”结尾，而下一个块以“abab”开头，overlap 将是“bababa”。您是否想计算中间的“abab”？也就是说，“abab”在“abababab”中出现了 2 次还是 3 次？上面的代码是 3。

但是对于没有前缀等于后缀的目标（例如“TIME_DATA”）不会出现这种歧义。例如，“ATIME_DATA”（“A”既是后缀又是前缀）可能会出现：它在“ATIME_DATATIME_DATA”中出现一次还是两次？如果上面的代码被分割成块，例如“...ATIME_DATATI”和“ME_DATA...”，则可以说“两次”。

如果您关心这一点，可以通过进行简短搜索来解决它，以确保粘贴在一起的片段中的匹配不会与靠近左侧块末尾或右侧块开头附近的匹配重叠.