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

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

问题是：

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

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

Answer 1

这是正确的行为吗？

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

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

你的例子的工作方式也许最好用一个比喻来说明——请多多包涵：

假设您请朋友进行一项实验。您希望他尽可能快地浏览一本书并用钢笔标记每一页。有两轮设置不同，你要为每一轮计时，然后比较哪一轮更快：

1. 在第一页打开书，标记它，然后翻页并标记接下来的页面。纯顺序处理。

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

哪些方法会更快？

我做错了吗？

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

1. 从书中剪下页面并将它们分类为 10 叠
2. 让十个人每人标记一叠

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

基本上这意味着任何进程的（理论上）加速只能与并行处理部分的速度一样快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，因为有几种记录在案的方法可以做到这一点，例如here 或 here。

结论

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

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