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

我正在尝试使用 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 

请您参考如下方法:

其实答案很简单...

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

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

但我必须这样做:

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

然后效果非常好。