IPython-IPyparallel ошибка импорта

Question

Я скопировал все коды в рабочий каталог на всех моих машинах с двигателями. И мой код:

my_test.py
my_startegy.py
main.py

Так, main.py будет работать на Client Machineкоды в main.py являются:

from ipyparallel import Client

import my_test
import my_strategy as strategy


class Beck_Test_Parallel(object):
    """
    """
    def __init__(self):
        self.rc = None
        self.dview = None

    def start_client(self, path):
        self.rc = Client(path)
        self.dview = self.rc[:]
        #self.dview.push(dict(
        #        Account=my_test.Account, 
        #        dataImport=my_test.dataImport
        #    ))

    def parallel_map(self, deal_function, accounts):
        import my_test
        return self.dview.map_sync(deal_function, accounts)

def create_accounts(time_list, account):
    accounts = []
    for index, time in enumerate(time_list):
        acc = my_test.Account(
                strategy.start, 
                strategy.end, 
                strategy.freq,
                strategy.universe_code, 
                strategy.capital_base, 
                strategy.short_capital, 
                strategy.benchmark, 
                strategy.self_defined
            )
        account.share_data(acc)
        acc.iniData2()
        acc.iniData3()
        acc.current_time = time
        acc.days_counts = index+1
        acc.dynamic_record['capital'] = acc.capital_base
        del acc.connect
        accounts.append(acc)
    return accounts

def let_us_deal(account):
    account =  strategy.handle_data(account) 
    print '   >>>', account.current_time
    return account


if __name__ == '__main__':
    account = my_test.Account(
            strategy.start,
            strategy.end,
            strategy.freq,
            strategy.universe_code,
            strategy.capital_base,
            strategy.short_capital,
            strategy.benchmark,
            strategy.self_defined
        )

    account.iniData()
    account.iniData2()
    account.iniData3()

    time_list = my_test.get_deal_time_list(account)

    accounts = parallel.create_accounts(time_list, account)

    back_test_parallel = parallel.Beck_Test_Parallel()

    back_test_parallel.start_client(
        '/home/fit/.ipython/profile_default/security/ipcontroller-client.json')

    back_test_parallel.dview.execute('import my_test')
    back_test_parallel.dview.execute('import my_strategy as strategy')
    # get the result
    result = back_test_parallel.parallel_map(let_us_deal, accounts)

    for acc in result.get():
        print acc.reselected_stocks, acc.current_time

И я импортировал my_test модуль в parallel_map() функция в классе Back_Test_Parallel и я также импортировал my_test модуль в back_test_parallel.dview.execute('import my_test'),

И соответствующие модули находятся в рабочем каталоге машины двигателя. Я скопировал ipcontroller-client.json а также ipcontroller-engine.json в рабочий каталог на engine machine,

Но когда он запускается, ошибка ImportError: No module named my_test, так как модуль my_test.py уже находится в рабочем каталоге. Это действительно заставило меня чувствовать разочарование!

---------------------------------------------------------------------------
CompositeError                            Traceback (most recent call last)
/home/fit/log/1027/back_test/main.py in <module>()
    119     import ipdb
    120     ipdb.set_trace()
--> 121     for acc in result.get():
    122         print acc.reselected_stocks, acc.current_time
    123 

/usr/local/lib/python2.7/dist-packages/ipyparallel/client/asyncresult.pyc in get(self, timeout)
    102                 return self._result
    103             else:
--> 104                 raise self._exception
    105         else:
    106             raise error.TimeoutError("Result not ready.")

CompositeError: one or more exceptions from call to method: let_us_deal
[0:apply]: ImportError: No module named my_test
[1:apply]: ImportError: No module named my_test

Кое-что о result:

In [2]: result
Out[2]: <AsyncMapResult: finished>
In [3]: type(result)
Out[3]: ipyparallel.client.asyncresult.AsyncMapResult

Обратите внимание, что когда он работает на одной машине с помощью ipcluster start -n 8работает нормально, без ошибок.
заранее спасибо

Answer 1

Я думаю, что мой CWD не в правильном каталоге. Таким образом, вы можете проверить свой CWD

>>> import os
>>> print(dview.apply_sync(os.getcwd).get())

Если он находится в неправильном каталоге, перед параллельными вычислениями вы можете установить правильный CWD, чтобы убедиться, что ipyparallel env находится в правильном рабочем каталоге:

>>> import os
>>> dview.map(os.chdir, ['/path/to/my/project/on/engine']*number_of_engines)
>>> print(dview.apply_sync(os.getcwd).get())

Вы также можете проверить название вашего двигателя по

>>> import socket
>>> print(dview.apply_sync(socket.gethostname))

И это работает отлично!

Answer 2

pip установить ipyparallel --upgrade