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)