Передача объекта параллельного кластера IPython в пользовательский класс для пакетного выполнения

Question

Я начинающий программист, пытающийся использовать python для научного программирования. Я думаю, что эти посты ( Как работать с интерактивно определенными классами в IPython.parallel? И пользовательском объекте параллельного push ipython) затрагивают аналогичную проблему, но для меня они бесполезны. Я хочу запустить свой код в виде скрипта (для планировщиков в очереди PBS или SGE) и не знаю, как бы я использовал укроп.

По сути, я пытаюсь использовать параллельный кластер Ipython для разделения вычислений, определенных в методе пользовательского класса.

Я хочу передать объект кластера в мой экземпляр пользовательского класса, а затем использовать кластер для разделения вычислений, которые работают с частями данных, определенными как член.

1. Запустив кластер используя ipcluster (/path/to/ipcontroller-client.json),
2. Тогда я хочу бежать, python test_parallel.py
3. Куда, test_parallel.py является

---

class Foo(object):
    def __init__(self):
        from numpy import arange
        self.data = arange(10)*10

    def A(self, y):
        print "in A:", y
        self.data[y]

    def parallelA(self, z, cl):
        print "in parallelA:", cl[:].map_sync(self.A, z)

    def serialA(self, z):
        print "in serialA:", map(self.A, z)

if __name__ == "__main__":

    from IPython.parallel import Client
    f = '/path/to/security/ipcontroller-client.json'
    c = Client(f)

    asdf = Foo()
    asdf.serialA([1, 3, 5])      ## works
    asdf.parallelA([1, 3, 5], c) ## doesn't work

---

Выход

---

$ ~/Projects/parcellation$ python test_parallel.py 
in serialA: in A: 1
in A: 3
in A: 5
[None, None, None]
in parallelA:
Traceback (most recent call last):
  File "test_parallel.py", line 24, in <module>
    asdf.parallelA([1, 3, 5], c) ## doesn't work
  File "test_parallel.py", line 11, in parallelA
    print "in parallelA:", cl[:].map_sync(self.A, z)
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/view.py", line 366, in map_sync
    return self.map(f,*sequences,**kwargs)
  File "<string>", line 2, in map
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/view.py", line 66, in sync_results
    ret = f(self, *args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/view.py", line 624, in map
    return pf.map(*sequences)
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/remotefunction.py", line 271, in map
    ret = self(*sequences)
  File "<string>", line 2, in __call__
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/remotefunction.py", line 78, in sync_view_results
    return f(self, *args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/remotefunction.py", line 243, in __call__
    ar = view.apply(f, *args)
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/view.py", line 233, in apply
    return self._really_apply(f, args, kwargs)
  File "<string>", line 2, in _really_apply
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/view.py", line 66, in sync_results
    ret = f(self, *args, **kwargs)
  File "<string>", line 2, in _really_apply
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/view.py", line 51, in save_ids
    ret = f(self, *args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/view.py", line 567, in _really_apply
    ident=ident)
  File "/usr/local/lib/python2.7/dist-packages/IPython/parallel/client/client.py", line 1263, in send_apply_request
    item_threshold=self.session.item_threshold,
  File "/usr/local/lib/python2.7/dist-packages/IPython/kernel/zmq/serialize.py", line 145, in pack_apply_message
    arg_bufs = flatten(serialize_object(arg, buffer_threshold, item_threshold) for arg in args)
  File "/usr/local/lib/python2.7/dist-packages/IPython/utils/data.py", line 30, in flatten
    return [x for subseq in seq for x in subseq]
  File "/usr/local/lib/python2.7/dist-packages/IPython/kernel/zmq/serialize.py", line 145, in <genexpr>
    arg_bufs = flatten(serialize_object(arg, buffer_threshold, item_threshold) for arg in args)
  File "/usr/local/lib/python2.7/dist-packages/IPython/kernel/zmq/serialize.py", line 89, in serialize_object
    buffers.insert(0, pickle.dumps(cobj, PICKLE_PROTOCOL))
cPickle.PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup __builtin__.instancemethod failed

---

Любая помощь в понимании, почему это не работает, и исправление, которое требует минимального изменения кода, было бы очень полезно.

Спасибо!

Answer 1

Я нашел решение:

class Foo(object):
    def __init__(self):
        from numpy import arange
        self.data = arange(10)*10

    @staticmethod
    def A(data, y):
        print "in A:", y ## doesn't produce an output
        return data[y]

    def parallelA(self, z, cl):
        print "in parallelA:", cl[:].map_sync(self.A, [self.data]*len(z), z)

if __name__ == "__main__":

    from IPython.parallel import Client
    f = '/path/to/security/ipcontroller-client.json'
    c = Client(f)

    asdf = Foo()
    asdf.parallelA([1, 3, 5], c)

Вывод, когда приведенный выше код выполняется:

$ python test_parallel.py 
in parallelA: [10, 30, 50]