python - 使用多处理池读取 CSV 比 CSV 阅读器花费的时间更长

Question

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

根据大小，模块决定是使用Multiprocessing pool 还是使用普通的CSV reader 读取文件。但从观察来看，多处理 比正常 CSV 读取 花费更长的时间，当测试大小为 100 MB 的文件的两种模式时。

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

这是我的代码:

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

Is this correct behaviour?

是的 - 它可能不是您所期望的，但它与您实现它的方式以及 multiprocessing 的工作方式是一致的。

Why Async mode is taking longer?

你的例子的工作方式也许最好用一个寓言来说明 - 请耐心等待:

假设您邀请您的 friend 参与一项实验。您希望他尽可能快地浏览一本书并用笔在每一页上做记号。有两轮具有不同的设置，您将为每一轮计时，然后比较哪一轮更快:

1. 打开书的第一页，做记号，然后翻页，在接下来的几页上做记号。纯顺序处理。

2. 分块处理这本书。为此，他应该逐页浏览这本书。那就是他应该首先列出页码作为起点，比如 1、10、20、30、40 等。然后对于每个 block ，他应该合上书，在起点页面上打开它，在下一个起点出现之前处理所有页面，关闭这本书，然后从头开始阅读下一部分。

哪种方法会更快？

Am I doing something wrong?

您认为这两种方法都花费了太长时间。您真正想要做的是让多个 人(进程)并行 进行标记。现在，对于一本书(就像一个文件)来说，这很困难，因为在任何时候只有一个人(过程)可以访问这本书(文件)。如果处理顺序无关紧要并且应该并行运行的是标记本身 - 而不是访问 - 仍然可以完成。所以新的方法是这样的:

1. 将书页剪下来，分成 10 叠
2. 让十个人每人标记一堆

这种方法肯定会加快整个过程。也许令人惊讶的是，虽然加速将小于 10 倍，因为第 1 步需要一些时间，而且只有一个人可以做到。那叫Amdahl's law [维基百科]:

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

直观上，加速只能来自并行处理的任务部分，所有顺序部分不受影响并且花费相同的时间，无论 p 是否被处理并行与否。

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

develop an application which should be able to process huge CSV files

解决方法如下:

1. 确定处理的哪些部分可以并行完成，即单独处理每个 block ，不按顺序
2. 按顺序读取文件，边读边将其分成 block
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，因为有几种记录的方法可以做到这一点，例如here或 here .

结论

根据每个文件所需的处理类型，顺序处理任何一个文件的方法很可能是最快的方法，原因很简单，因为处理部分不会从并行完成中获益太多。由于例如，您可能仍会逐 block 处理它。内存限制。如果是这种情况，您可能不需要多处理。

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