使用多处理池读取 CSV 比 CSV reader 花费的时间更长

Question

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

根据大小，模块决定是使用 Multiprocessing pool 还是使用正常 CSV reader 读取文件。但是根据观察，multi processing 在测试大小为 100 MB 的文件的两种模式时比正常 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

问题是：

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

注意： 删除了不必要的 functions/lines 以避免代码变得复杂

Answer 1

Is this correct behaviour?

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

Why Async mode is taking longer?

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

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

打开书的第一页，做记号，然后翻页，在后面的页上做记号。纯顺序处理。
分块处理这本书。为此，他应该运行一页一页地翻阅这本书。那就是他应该首先列出页码作为起点，比如 1、10、20、30、40 等。然后对于每个块，他应该合上书，在起点页面上打开它，在下一个起点出现之前处理所有页面，关闭这本书，然后重新开始下一个章节。

以下哪种方法会更快？

Am I doing something wrong?

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

把书页剪下来，然后整理成 10 叠
请十个人每人标记一叠

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

本质上它的意思是任何进程的（理论上的）加速只能和并行处理部分一样快 p 相对于零件的顺序处理时间（p/s）。

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

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

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

解决方法如下：

确定 处理的哪一部分 可以并行完成，即单独处理每个块并且不按顺序处理
按顺序读取文件，边读边将其分成块
使用多处理运行多个处理步骤并行

像这样：

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 or .

结论

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

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

使用多处理池读取 CSV 比 CSV reader 花费的时间更长

Reading CSV with multiprocessing pool is taking longer than CSV reader

python

csv

python-multiprocessing