使用多处理池读取CSV所需的时间比CSV读取器长

Question

根据我们客户的要求,我必须开发一个应该能够处理大量CSV文件的应用程序.文件大小可以在10 MB - 2GB的范围内.

根据大小,模块决定是Multiprocessing pool使用普通文件还是使用普通文件来读取文件CSV reader.但是从观察开始,在测试大小为100 MB的文件的两种模式时,multi processing需要比平时更长的时间CSV reading.

这是正确的行为吗？或者我做错了什么？

这是我的代码:

def set_file_processing_mode(self, fpath):
   """ """
   fsize = self.get_file_size(fpath)
   if fsize > FILE_SIZE_200MB:
      self.read_in_async_mode = True
   else:
      self.read_in_async_mode = False

def read_line_by_line(self, filepath):
    """Reads CSV line by line"""
    with open(filepath, 'rb') as csvin:
        csvin = csv.reader(csvin, delimiter=',')
        for row in iter(csvin):
          yield row

def read_huge_file(self, filepath):
    """Read file in chunks"""
    pool = mp.Pool(1)
    for chunk_number in range(self.chunks): #self.chunks = 20
        proc = pool.apply_async(read_chunk_by_chunk, 
                        args=[filepath, self.chunks, chunk_number])
        reader = proc.get()
        yield reader
    pool.close()
    pool.join()

def iterate_chunks(self, filepath):
    """Read huge file rows"""
    for chunklist in self.read_huge_file(filepath):
        for row in chunklist:
            yield row
@timeit #-- custom decorator
def read_csv_rows(self, filepath):
    """Read CSV rows and pass it to processing"""
    if self.read_in_async_mode:
        print("Reading in async mode")
        for row in self.iterate_chunks(filepath):
            self.process(row)
    else:
        print("Reading in sync mode")
        for row in self.read_line_by_line(filepath):
            self.process(row)

def process(self, formatted_row):
    """Just prints the line"""
    self.log(formatted_row)

def read_chunk_by_chunk(filename, number_of_blocks, block):
  '''
  A generator that splits a file into blocks and iterates
  over the lines of one of the blocks.
  '''
  results = []
  assert 0 <= block and block < number_of_blocks
  assert 0 < number_of_blocks
  with open(filename) as fp :
    fp.seek(0,2)
    file_size = fp.tell()
    ini = file_size * block / number_of_blocks
    end = file_size * (1 + block) / number_of_blocks
    if ini <= 0:
        fp.seek(0)
    else:
        fp.seek(ini-1)
        fp.readline()
    while fp.tell() < end:
        results.append(fp.readline())
  return results

if __name__ == '__main__':
    classobj.read_csv_rows(sys.argv[1])    

这是一个测试:

$ python csv_utils.py "input.csv"
Reading in async mode
FINISHED  IN 3.75 sec
$ python csv_utils.py "input.csv"
Reading in sync mode
FINISHED  IN 0.96 sec

问题是:

为什么异步模式需要更长的时间？

注意: 删除不必要的函数/行以避免代码的复杂性

Answer 1

这是正确的行为吗？

是的 - 它可能不是您所期望的,但它与您实施它的方式以及如何multiprocessing工作一致.

为什么异步模式需要更长的时间？

你的榜样的工作方式或许最能说明一个寓言 - 请跟我说:

假设您要求您的朋友参与实验.你希望他能够通过一本书,用笔尽可能快地用笔标记每一页.有两轮具有不同的设置,你将每轮时间,然后比较哪一个更快:

1. 在第一页上打开书籍,标记它,然后翻页并在出现时标记以下页面.纯顺序处理.

2. 大块处理这本书.为此,他应该按照大块的方式浏览这本书的页面.那就是他应该首先列出一个页码列表作为起点,比如1,10,20,30,40等.然后对于每个块,他应该关闭书,在页面上打开它作为起点,处理在下一个起点之前的所有页面出现,关闭书籍,然后重新开始下一个块.

以下哪种方法会更快？

难道我做错了什么？

你认为这两种方法都需要很长时间.你真正想做的是让多个人(进程)并行进行标记.现在有一本书(与文件一样)很难,因为,只有一个人(进程)可以在任何一点访问该书(文件).如果处理顺序无关紧要并且标记本身 - 而不是访问 - 应该并行运行,那么它仍然可以完成.所以新方法是这样的:

1. 从页面中删除页面并将它们分成10个堆栈
2. 让十个人分别标记一个堆栈

这种方法肯定会加速整个过程.也许令人惊讶的是,加速速度将低于10倍,因为步骤1需要一些时间,而且只有一个人可以做到.这就是所谓的阿姆达尔定律 [维基百科]:

本质上意味着任何过程的(理论上)加速只能与并行处理部分p相对于部件的顺序处理时间(p/s)的速度降低一样快.

直观地说,加速只能来自并行处理的任务部分,所有顺序部分都不受影响并占用相同的时间,无论p是否并行处理.

也就是说,在我们的例子中,显然加速只能来自第2步(由多人并行标记页面),因为步骤1(撕毁书籍)显然是连续的.

开发一个应该能够处理巨大的CSV文件的应用程序

以下是如何处理此问题:

1. 确定可以并行完成处理的哪个部分,即分别处理每个块并且不按顺序处理
2. 按顺序读取文件,随时将其拆分为块
3. 使用多处理并行运行多个处理步骤

像这样的东西:

def process(rows):
    # do all the processing
    ...
    return result

if __name__ == '__main__':
    pool = mp.Pool(N) # N > 1
    chunks = get_chunks(...)
    for rows in chunks:
       result += pool.apply_async(process, rows)
    pool.close()
    pool.join() 

我没有get_chunks在这里定义,因为有几种记录的方法可以做到这一点,例如这里或这里.

结论

根据每个文件所需的处理类型,处理任何一个文件的顺序方法很可能是最快的方法,因为处理部分并不是通过并行完成而获得的.由于例如内存限制,您可能仍然最终会按块处理它.如果是这种情况,您可能不需要多处理.

如果您有多个可以并行处理的文件,则多处理是一种非常好的方法.它的工作方式与上面显示的相同,其中块不是行而是文件名.