Distribution de graphes à travers les nœuds de cluster

Question

Je progresse bien avec Dask.delayed. En tant que groupe, nous avons décidé de consacrer plus de temps à travailler avec les graphiques en utilisant Dask.

J'ai une question à propos de distribution. Je vois le comportement suivant sur notre cluster. Je démarre par exemple 8 travailleurs sur chacun des 8 nœuds avec chacun 4 threads, disons/alors client.compute 8 graphiques pour créer les données simulées pour le traitement ultérieur. Je veux que les 8 ensembles de données soient générés un par nœud. Cependant, ce qui semble se produire est, non déraisonnable, les huit fonctions sont exécutées sur les deux premiers nœuds. Les calculs suivants sont exécutés sur les premier et deuxième noeuds. Par conséquent, je vois un manque de mise à l'échelle. Au fil du temps, les autres nœuds disparaissent de la page des travailleurs de diagnostic. Est-ce prévu? Donc, je veux d'abord distribuer les fonctions de création de données par nœud. Donc, quand je veux calculer les graphiques, maintenant je fais:

if nodes is not None: 
    print("Computing graph_list on the following nodes: %s" % nodes) 
    return client.compute(graph_list, sync=True, workers=nodes, **kwargs) 
else: 
    return client.compute(graph_list, sync=True, **kwargs) 

Cela semble configuré correctement: la barre de progression de diagnostic montre que mes données fonctions de création sont en mémoire mais ils ne démarrent pas. Si les noeuds sont omis, le calcul se déroule comme prévu. Ce problème se produit à la fois sur le cluster et sur mon bureau.

Plus d'informations: en regardant le journal du planificateur, je vois des échecs de communication.

more dask-ssh_2017-09-04_09\:52\:09/dask_scheduler_sand-6-70\:8786.log 
distributed.scheduler - INFO - ----------------------------------------------- 
distributed.scheduler - INFO - Scheduler at: tcp://10.143.6.70:8786 
distributed.scheduler - INFO -  bokeh at:    0.0.0.0:8787 
distributed.scheduler - INFO -  http at:    0.0.0.0:9786 
distributed.scheduler - INFO - Local Directory: /tmp/scheduler-ny4ev7qh 
distributed.scheduler - INFO - ----------------------------------------------- 
distributed.scheduler - INFO - Register tcp://10.143.6.73:36810 
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.6.73:36810 
distributed.scheduler - INFO - Register tcp://10.143.6.71:46656 
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.6.71:46656 
distributed.scheduler - INFO - Register tcp://10.143.7.66:42162 
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.7.66:42162 
distributed.scheduler - INFO - Register tcp://10.143.7.65:35114 
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.7.65:35114 
distributed.scheduler - INFO - Register tcp://10.143.6.70:43208 
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.6.70:43208 
distributed.scheduler - INFO - Register tcp://10.143.7.67:45228 
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.7.67:45228 
distributed.scheduler - INFO - Register tcp://10.143.6.72:36100 
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.6.72:36100 
distributed.scheduler - INFO - Register tcp://10.143.7.68:41915 
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.7.68:41915 
distributed.scheduler - INFO - Receive client connection: 5d1dab2a-914e-11e7-8bd1-180373ff6d8b 
distributed.scheduler - INFO - Worker 'tcp://10.143.6.71:46656' failed from closed comm: Stream is clos 
ed 
distributed.scheduler - INFO - Remove worker tcp://10.143.6.71:46656 
distributed.scheduler - INFO - Removed worker tcp://10.143.6.71:46656 
distributed.scheduler - INFO - Worker 'tcp://10.143.6.73:36810' failed from closed comm: Stream is clos 
ed 
distributed.scheduler - INFO - Remove worker tcp://10.143.6.73:36810 
distributed.scheduler - INFO - Removed worker tcp://10.143.6.73:36810 
distributed.scheduler - INFO - Worker 'tcp://10.143.6.72:36100' failed from closed comm: Stream is clos 
ed 
distributed.scheduler - INFO - Remove worker tcp://10.143.6.72:36100 
distributed.scheduler - INFO - Removed worker tcp://10.143.6.72:36100 
distributed.scheduler - INFO - Worker 'tcp://10.143.7.67:45228' failed from closed comm: Stream is clos 
ed 
distributed.scheduler - INFO - Remove worker tcp://10.143.7.67:45228 
distributed.scheduler - INFO - Removed worker tcp://10.143.7.67:45228 
(arlenv) [hpccorn1@login-sand8 performance]$ 

Cela soulève-t-il des causes possibles?

Merci, Tim

Answer 1

Comment Dask choisit de répartir les tâches aux travailleurs est complexe, et prend en compte de nombreux problèmes tels que l'équilibrage de charge, le transfert de données, les contraintes de ressources, etc .. Il peut être difficile de raisonner sur l'endroit où les choses finiront sans un exemple concret et simple.

Une chose que vous pouvez essayer est de soumettre tous vos calculs à la fois, cela permet au planificateur de prendre des décisions un peu plus intelligentes plutôt que de voir les choses une à la fois.

Ainsi, vous pouvez essayer de remplacer le code comme ceci:

futures = [client.compute(delayed_value) for delayed_value in L] 
wait(futures) 

avec le code suivant

futures = client.compute(L) 
wait(futures) 

Mais honnêtement je ne donne cette chance de 30% de résoudre votre problème. Il est difficile de savoir ce qui se passe sans plonger beaucoup plus profondément dans votre problème. Si vous pouvez fournir un très simple exemple de code reproductible alors ce serait mieux.