python - 使用多处理在 Python 中读取多个 HDF5 文件

Question

我正在尝试使用 PyTables 和 multiprocessing 读取一堆 HDF5 文件(“一堆”意味着 N > 1000 个文件)。基本上，我创建了一个类来读取数据并将其存储在 RAM 中；它在顺序模式下工作得非常好，我想将它并行化以获得一些性能。

我现在尝试了一种虚拟方法，为我的类创建了一个新方法 flatten() 来并行化文件读取。以下示例是我正在尝试执行的操作的简化示例。 listf 是包含要读取的文件名的字符串列表，nx 和 ny 是我要读取的数组的大小文件:

import numpy as np
import multiprocessing as mp
import tables

class data:
    def __init__(self, listf, nx, ny, nproc=0):
        self.listinc = []
        for i in range(len(listf)):
             self.listinc.append((listf[i], nx, ny))

    def __del__(self):
        del self.listinc

    def get_dsets(self, tuple_inc):
        listf, nx, ny = tuple_inc
        x = np.zeros((nx, ny))
        f = tables.openFile(listf)
        x = np.transpose(f.root.x[:ny,:nx])
        f.close()
        return(x)

    def flatten(self):
        nproc = mp.cpu_count()*2

        def worker(tasks, results):
            for i, x in iter(tasks.get, 'STOP'):
                print i, x
                results.put(i, self.get_dsets(x))

        tasks   = mp.Queue()
        results = mp.Queue()
        manager = mp.Manager()
        lx      = manager.list()

        for i, out in enumerate(self.listinc):
            tasks.put((i, out))

        for i in range(nproc):
            mp.Process(target=worker, args=(tasks, results)).start()

        for i in range(len(self.listinc)):
            j, res = results.get()
            lx.append(res)

        for i in range(nproc):
            tasks.put('STOP')

我尝试了不同的方法(包括，就像我在这个简单的例子中所做的那样，使用 manager 来检索数据)但我总是得到一个 TypeError: an integer is required.

我不使用 ctypes 数组，因为我真的不需要共享数组(我只想检索我的数据)并且在检索数据后，我想用 NumPy 来玩弄它。

任何想法、提示或帮助将不胜感激!

编辑:我得到的完整错误如下:

Process Process-341:
Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python2.7/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "/home/toto/test/rd_para.py", line 81, in worker
    results.put(i, self.get_dsets(x))
  File "/usr/lib/python2.7/multiprocessing/queues.py", line 101, in put
    if not self._sem.acquire(block, timeout):
TypeError: an integer is required

Answer 1

其实答案很简单...

在 worker 中，因为它是我检索的元组，所以我不能这样做:

result.put(i, self.get_dsets(x))

但我必须这样做:

result.put((i, self.get_dsets(x)))

然后效果非常好。