python - 优化此 python 日志解析代码

Question

对于 4.2 GB 输入文件，此代码在我的笔记本电脑上的运行时间为 48 秒。输入文件以制表符分隔，每个值都出现在引号中。每条记录以换行符结尾，例如'"val1"\t"val2"\t"val3"\t..."valn"\n'

我已经尝试使用具有 10 个线程的多处理:一个用于排队行，8 个用于解析单个行并填充输出队列，一个用于将输出队列缩减为如下所示的 defaultdict，但代码花费了 300 秒运行，比以下时间长 6 倍以上:

from collections import defaultdict
def get_users(log):
    users = defaultdict(int)
    f = open(log)
    # Read header line
    h = f.readline().strip().replace('"', '').split('\t')
    ix_profile = h.index('profile.type')
    ix_user = h.index('profile.id')
    # If either ix_* is the last field in h, it will include a newline. 
    # That's fine for now.
    for (i, line) in enumerate(f): 
        if i % 1000000 == 0: print "Line %d" % i # progress notification

        l = line.split('\t')
        if l[ix_profile] != '"7"': # "7" indicates a bad value
            # use list slicing to remove quotes
            users[l[ix_user][1:-1]] += 1 

    f.close()
    return users

我已经通过从 for 循环中删除除 print 语句之外的所有内容来检查我是否不受 I/O 限制。该代码在 9 秒内运行，我认为这是该代码运行速度的下限。

我有很多这样的 5 GB 文件要处理，所以即使在运行时间上有很小的改进(我知道，我可以删除打印件!)也会有所帮助。我运行的机器有 4 个内核，所以我不禁想知道是否有办法让多线程/多进程代码比上面的代码运行得更快。

更新:

我重写了多处理代码如下:

from multiprocessing import Pool, cpu_count
from collections import defaultdict

def parse(line, ix_profile=10, ix_user=9):
    """ix_profile and ix_user predetermined; hard-coding for expedience."""
    l = line.split('\t')
    if l[ix_profile] != '"7"':
        return l[ix_user][1:-1]

def get_users_mp():
    f = open('20110201.txt')
    h = f.readline() # remove header line
    pool = Pool(processes=cpu_count())
    result_iter = pool.imap_unordered(parse, f, 100)
    users = defaultdict(int)
    for r in result_iter:
        if r is not None:
            users[r] += 1
    return users

它运行时间为 26 秒，速度提高了 1.85 倍。还不错，但有 4 个内核，没有我希望的那么多。

Answer 1

使用正则表达式。

测试确定该过程的昂贵部分是对 str.split() 的调用。可能必须为每一行构建一个列表和一堆字符串对象是昂贵的。

首先，您需要构造一个正则表达式来匹配该行。像这样的东西:

expression = re.compile(r'("[^"]")\t("[^"]")\t')

如果您调用 expression.match(line).groups()，您将把前两列提取为两个字符串对象，您可以直接对它们进行逻辑运算。

现在假设感兴趣的两列是前两列。如果不是，您只需调整正则表达式以匹配正确的列。您的代码检查标题以查看列所在的位置。您可以基于此生成正则表达式，但我猜这些列实际上总是位于同一位置。只需验证它们仍然存在并在行上使用正则表达式即可。

编辑

从集合导入默认字典导入重新

def get_users(log):
    f = open(log)
    # Read header line
    h = f.readline().strip().replace('\'', '').split('\t')
    ix_profile = h.index('profile.type')
    ix_user = h.index('profile.id')

    assert ix_user < ix_profile

此代码假定用户位于个人资料之前

    keep_field = r'"([^"]*)"'

此正则表达式将捕获单个列

    skip_field = r'"[^"]*"'

此正则表达式将匹配列，但不会捕获结果。 (注意没有括号)

    fields = [skip_field] * len(h)
    fields[ix_profile] = keep_field
    fields[ix_user] = keep_field

为所有字段创建一个列表，只保留我们关心的字段

    del fields[max(ix_profile, ix_user)+1:]

把我们关心的后面的字段都去掉(匹配需要时间，我们不关心)

    regex = re.compile(r"\t".join(fields))

实际生成正则表达式。

    users = defaultdict(int)
    for line in f:
        user, profile = regex.match(line).groups()

把两个值都拉出来，做逻辑

        if profile != "7": # "7" indicates a bad value
            # use list slicing to remove quotes
            users[user] += 1 

    f.close()
    return users