3

PyTables我正在尝试使用and读取一堆 HDF5 文件（“一堆”表示 N > 1000 个文件）multiprocessing。基本上，我创建了一个类来读取我的数据并将其存储在 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')

我尝试了不同的方法（包括，就像我在这个简单示例中所做的那样，使用 amanager来检索数据），但我总是得到一个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
4

1 回答 1

0

答案其实很简单……

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

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

但我必须这样做：

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

然后效果很好。

于 2013-01-17T16:35:17.013 回答