Erreur d'importation IPython-IPyparallel

Question

J'ai copié tous les codes dans le répertoire de travail de toutes mes machines-moteurs. Et mon code sont:

my_test.py 
my_startegy.py 
main.py 

Ainsi, main.py sera couru sur Client Machine, les codes à main.py sont:

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 

et moi-même avons importé le module my_test en parallel_map() fonction en classe Back_Test_Parallel et j'ai aussi module my_test importé dans back_test_parallel.dview.execute('import my_test').

Et les modules correspondants sont dans le répertoire de travail de la machine du moteur. J'ai copié les ipcontroller-client.json et ipcontroller-engine.json dans le répertoire de travail sur engine machine.

Mais quand il s'exécute, l'erreur est ImportError: No module named my_test, puisque le module my_test.py est déjà dans le répertoire de travail. Cela m'a vraiment frustré!

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

quelque chose result:

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

Notez que, quand il fonctionne sur une seule machine en utilisant ipcluster start -n 8, il fonctionne très bien, sans aucune erreur.
Merci d'avance

Answer 1

Je pense que mon CWD n'est pas dans le bon répertoire. Ainsi, vous pouvez vérifier votre CWD

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

Si elle est dans un mauvais répertoire, avant le calcul parallèle, vous pouvez définir le droit CWD pour vous assurer que vous ipyparallel env est dans le bon répertoire de travail:

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

Vous pouvez également vérifier le nom de votre moteur par

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

Et ça marche bien!