python使用多進程來過濾海量文件

Question

我想並行化一個文件過濾操作，其中每個過濾器是一個大的正則表達式，因此整個事情需要時間來運行。該文件本身大約100GB。單進程的版本是這樣的：

def func(line): 
    # simple function as an example 
    for i in range(10**7): 
     pass 
    return len(line) % 2 == 0 


with open('input.txt') as in_sr, open('output.txt', 'w') as out_sr: 
    for line in input: 
     if func(line): 
      out_sr.write(line) 

我嘗試使用multiprocessing的imap但給人ValueError: I/O operation on closed file.我認爲迭代被複制到每一個過程，但不是所有的進程有處理打開。

有沒有辦法做到這一點使用multiprocessing，最好是利用池？

Answer 1

我沒有錯誤運行下面的代碼。確保您不要在with聲明之外撥打in_sr和out_sr。

from multiprocessing import Pool 

def func(line): 
    # simple function as an example 
    for i in xrange(10**7): 
     pass 
    return len(line) % 2 == 0, line 

def main(): 
    with open('input.txt','r') as in_sr, open('output.txt', 'w') as out_sr: 
     pool = Pool(processes=4) 
     for ret,line in pool.imap(func, in_sr, chunksize=4): 
      if ret: 
       out_sr.write(line) 
     pool.close() 

if __name__ == '__main__': 
    main() 

Answer 2

的代碼與此類似：

def func(line): 
    ... 

if __name__ == '__main__': 

    from multiprocessing import Pool 
    from itertools import tee, izip 

    pool = Pool(processes=4) 

    with open('input.txt') as in_sr, open('output.txt', 'w') as out_sr: 
     lines1, lines2 = tee(in_sr) 
     for line, flag in izip(lines1, pool.imap(func, lines2)): 
      if flag: 
       out_sr.write(line)