Ошибка импорта 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 находится не в правильном каталоге. Таким образом, вы можете проверить УХО

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

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

>>> 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)) 

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