Python，并行处理一个大文本文件

Question

数据文件(SAM文件)中的样本记录:

M01383  0  chr4  66439384  255  31M  *  0  0  AAGAGGA GFAFHGD  MD:Z:31 NM:i:0
M01382  0  chr1  241995435  255 31M  *  0  0  ATCCAAG AFHTTAG  MD:Z:31 NM:i:0
......

• 数据文件是逐行的
• 数据文件的大小从 1G - 5G 不等。

我需要逐行遍历数据文件中的记录，从每一行中获取特定值(例如第 4 个值，66439384)，并将该值传递给另一个函数进行处理。然后一些结果计数器将被更新。

基本的工作流程是这样的:

# global variable, counters will be updated in search function according to the value passed. 
counter_a = 0    
counter_b = 0
counter_c = 0

open textfile:
    for line in textfile:
        value = line.split()[3]
        search_function(value)    # this function takes abit long time to process

def search_function (value):
    some conditions checking:
        update the counter_a or counter_b or counter_c

单进程代码，1.5G左右的数据文件，跑完一个数据文件的所有记录，耗时约20小时。我需要更快的代码，因为有超过 30 个此类数据文件。

我想并行处理N个chunk中的数据文件，每个chunk都会执行上面的workflow，同时更新全局变量(counter_a, counter_b, counter_c)。但我不知道如何在代码中实现这一点，或者这是否可行。

我可以访问一台服务器机器:24 个处理器和大约 40G RAM。

有人可以帮忙吗？非常感谢。

Answer 1

最简单的方法可能是用您现有的代码一次完成所有 30 个文件——仍然需要一整天，但您可以一次完成所有文件。 (即“9个月9个宝宝”容易，“1个月1个宝宝”难)

如果您真的想更快地完成单个文件，这将取决于您的计数器实际更新的方式。如果几乎所有工作都只是分析值(value)，您可以使用多处理模块卸载它:

import time
import multiprocessing

def slowfunc(value):
    time.sleep(0.01)
    return value**2 + 0.3*value + 1

counter_a = counter_b = counter_c = 0
def add_to_counter(res):
    global counter_a, counter_b, counter_c
    counter_a += res
    counter_b -= (res - 10)**2
    counter_c += (int(res) % 2)

pool = multiprocessing.Pool(50)
results = []

for value in range(100000):
    r = pool.apply_async(slowfunc, [value])
    results.append(r)

    # don't let the queue grow too long
    if len(results) == 1000:
        results[0].wait()

    while results and results[0].ready():
        r = results.pop(0)
        add_to_counter(r.get())

for r in results:
    r.wait()
    add_to_counter(r.get())

print counter_a, counter_b, counter_c

这将允许 50 个 slowfunc 并行运行，因此不需要 1000 秒 (=100k*0.01s)，而是需要 20 秒 (100k/50)*0.01s 才能完成。如果您可以像上面那样将您的函数重组为“slowfunc”和“add_to_counter”，您应该能够获得 24 倍的加速。