<jupyter_start><jupyter_text>2A.i - Jupyter et calcul distribué[Jupyter](http://jupyter.org/) a été découpé en plusieurs extensions comme [ipyparallel](http://ipyparallel.readthedocs.io/en/latest/) qui permet de distribuer un calcul sur plusieurs processus. Ce notebook montre comment faire sur une seule machine.<jupyter_code>from jyquickhelper import add_notebook_menu
add_notebook_menu()<jupyter_output><empty_output><jupyter_text>Avec de commencer, quelques mots de vocabulaire :* [thread](http://fr.wikipedia.org/wiki/Thread_%28informatique%29) : il est possible de paralléliser un traitement au sein d'un même programme ou processus, c'est un *thread*. Les threads ont accès aux mêmes variables.* [processus](http://fr.wikipedia.org/wiki/Processus_%28informatique%29) : c'est un programme, chaque processus a sa propre mémoire non partagée avec d'autres processus.  Une fontion utile pour récupérer les processus qui tournentVoir [psutil](https://pythonhosted.org/psutil/).<jupyter_code>import psutil

def find_process(name):
    for proc in psutil.process_iter():
        try: pinfo = proc.as_dict(attrs=['pid', 'name'])
        except psutil.NoSuchProcess: pass
        else:
            if name in pinfo["name"]:
                return pinfo, proc
    return None, None
i, p = find_process('ipcluster')
i, p<jupyter_output><empty_output><jupyter_text>Calcul distribuéQuelques liens :* [Overview and getting started with ipyrallel](https://ipyparallel.readthedocs.io/en/latest/)La distribution d'un programme implique l'exécution en parallèle de morceaux de programme, sur la même machine ou sur des machines différentes. Dans le cas de cette séance, ce sera sur la même machine mais le principe reste le même : une machine centrale (serveur) envoie des données et le traitement associé sur d'autres machines (clients). Pour que les clients comprennent qu'ils doivent exécuter un programme, ils doivent continuellement attendre qu'on leur envoie des instructions : un client est lancé puis attend des instructions. Pour lancer des clients localement sur cette machine, on doit aller dans le répertoire :<jupyter_code>import os,sys
if hasattr(sys, 'real_prefix'):
    exe = sys.real_prefix
else:
    exe = sys.base_exec_prefix
f = os.path.join(exe, "Scripts")
f
[ _ for _ in os.listdir(f) if "cluster" in _ ]<jupyter_output><empty_output><jupyter_text>Il faut exécuter l'instruction ``ipcluster start -n 2`` depuis une ligne de commande. Celle-ci vient avec le module [ipyparallel](https://ipyparallel.readthedocs.org/en/latest/) ``2`` signifie deux clients puis on vérifie que ceux-ci sont bien en train de s'exécuter. Sous Windows, il faut ouvrir une fenêtre de commande avec la commande [cmd](http://windows.microsoft.com/fr-fr/windows/command-prompt-faq1TC=windows-8).<jupyter_code># ipcluster start -n 2<jupyter_output><empty_output><jupyter_text>On obtient quelque chose similaire à ceci :<jupyter_code>from pyquickhelper.helpgen import NbImage
NbImage("cmdipc.png")<jupyter_output><empty_output><jupyter_text>On utilise une astuce pour le lancer depuis un noteboook :<jupyter_code>if find_process("ipcluster")[0] is None:
    print("start ipcluster")
    from pyquickhelper.loghelper import run_cmd
    if sys.platform.startswith("win"):
        cmd = os.path.join(f, "ipcluster")
    else:
        cmd = "ipcluster"
    cmd += " start -n 2"
    run_cmd(cmd, wait=False)
else:
    print("déjà démarré", find_process("ipcluster"))<jupyter_output>start ipcluster<jupyter_text>On attend que le processus démarre.<jupyter_code>import time
time.sleep(5)<jupyter_output><empty_output><jupyter_text>On vérifie que les deux clients sont accessibles depuis de notebook :<jupyter_code>from ipyparallel import Client
clients = Client()
clients.block = True  # use synchronous computations
print(clients.ids)    # on s'attend à deux clients<jupyter_output>[0, 1]<jupyter_text>Tout va bien. L'ensemble fonctionne comme le dessin ci-dessous :<jupyter_code>from IPython.core.display import Image
Image("http://ipython.org/ipython-doc/stable/_images/wideView.png", width=300)<jupyter_output><empty_output><jupyter_text>Exemple simple On teste l'exemple du [tutoriel](https://ipyparallel.readthedocs.io/en/latest/intro.htmlgetting-started) :<jupyter_code>clients[:].apply_sync(lambda : "Hello, World")<jupyter_output><empty_output><jupyter_text>Et un autre pour paralléliser l'exécution d'une fonction : [map_sync](https://ipyparallel.readthedocs.io/en/latest/multiengine.htmlparallel-map).<jupyter_code>def addition(m):
    return m + 1
parallel_result = clients[:].map_sync(addition, range(32))
parallel_result[:5]<jupyter_output><empty_output><jupyter_text>Exemple plus subtile On crée une autre fonction plus complexe qu'on souhaite paralléliser :<jupyter_code>def inverse_matrice(m) :
    return numpy.linalg.inv ( m )<jupyter_output><empty_output><jupyter_text>Ensuite, on créer 100 matrices à inverser :<jupyter_code>import numpy
ms = [ numpy.random.random ( (5,5) )  for i in range(0,10) ]<jupyter_output><empty_output><jupyter_text>On fait de même pour paralléliser la fonction. Selon les différentes installation, cela provoaque une erreur ou cela n'aboutit pas.<jupyter_code># mat = clients[:].map_sync(inverse_matrice, ms)<jupyter_output><empty_output><jupyter_text>Car le module ``numpy`` a été importé dans ce notebook mais il ne l'a pas été sur les processus distribués :<jupyter_code>with clients[:].sync_imports():
    import numpy<jupyter_output>importing numpy on engine(s)<jupyter_text>On réessaye :<jupyter_code>mat = clients[:].map_sync(inverse_matrice, ms)
mat[:2]<jupyter_output><empty_output><jupyter_text>Autre écriture et mesure de temps Puis on crée une [view](https://ipyparallel.readthedocs.io/en/latest/api/ipyparallel.html?ipyparallel.LoadBalancedView) *load balanced* :<jupyter_code>view = clients.load_balanced_view()<jupyter_output><empty_output><jupyter_text>Qu'on utilise pour dispatcher les résultats sur plusieurs processus :<jupyter_code>results = view.map(inverse_matrice, ms)
len(results)<jupyter_output><empty_output><jupyter_text>On recommence à exécuter le code suivant :<jupyter_code>results = view.map(inverse_matrice, ms)
len(results)<jupyter_output><empty_output><jupyter_text>On compare avec un traitement non distribué :<jupyter_code>%timeit list(map(inverse_matrice, ms))
%timeit view.map(inverse_matrice, ms)<jupyter_output>1 loop, best of 3: 294 ms per loop<jupyter_text>Le processus distribué est plus long : cela prend du temps de communiquer les matrices depuis ce notebook vers les *engines*. Ce temps est manifestement plus long. C'est pourquoi on distribue généralement des processus dont le calcul est significativement plus lent que le temps de communication. Selon ce schéma, on préfère envoyer des noms de fichiers au processus pour ce recevoir des résultats courts. Pour distribuer, il faut faire attention au ratio communication/calcul. La fonction suivante ne reçoit rien, fait beaucoup de calcul et retourne une matrice 10x10.<jupyter_code>def average_random_matrix(i):
    mean = None
    for n in range(0,100000):
        m = numpy.random.random ( (10,10) )
        if mean is None : mean = m
        else : mean += m
    return mean / n
%timeit list(map(average_random_matrix, range(0,10)))
%timeit view.map(average_random_matrix, range(0,10))<jupyter_output>1 loop, best of 3: 3.23 s per loop<jupyter_text>La distribution commence à montrer un gain de temps. Elle ne devient généralement intéressante que si un traitement numérique dure au moins quelques secondes.  Autres options pour distribuer les calculs IPython n'est pas le seul module à proposer un méchanisme de parallélisation. Un problème récurrent et souvent très agaçant est l'**erreur qui oblige à tout recommencer depuis le depuis**. On parallélise pour gagner du temps et une erreur se produit au bout de trois heures à cause d'une division par zéro par exemple. Il existe des modules qui sauvent les résultats intermédiaires et ne recommencent pas les calculs déjà effectués :* [joblib](http://pythonhosted.org//joblib/index.html) : très simple d'utilisation, le module parallélise des tâches sur plusieurs processus sans que vous ayez à vous soucier des échanges entre processus.* [luigi](https://github.com/spotify/luigi) : c'est un système pour les long [workflow](http://en.wikipedia.org/wiki/Workflow) (plusieurs heures ou jours), il permet de voir visuellement l'avancement.Il existe d'autres options comme [pyina](https://pypi.python.org/pypi/pyina). Une des premières solutions capable de faire cela était [RPyC](http://rpyc.readthedocs.org/en/latest/). Lorsqu'on veut distribuer sur plusieurs machines, il faut faire attention à :* Les machines sont-elles sur un réseau local ou distantes (Azure, Amazon, ...) ?* Les communications doivent-elles encryptées ?Quelques pointeurs :* [An introduction to 0mq](http://nichol.as/zeromq-an-introduction)* [Using IPython for parallel computing](http://ipython.org/ipython-doc/stable/parallel/index.html)* [mpi4py](http://pythonhosted.org//mpi4py/)  Un peu plus sur les threadsLe module [threading](https://docs.python.org/3.4/library/threading.html) propose une interface pour créer des threads. C'est utile pour exécuter des fonctions en parallèle. Toutefois, comme le langage Python n'est pas un langage multithreadé (voir [Global Interpreter Lock](http://en.wikipedia.org/wiki/Global_Interpreter_Lock), il ne sera pas plus rapide du point de vue vitesse de calcul. La fonction suivante montre comment à partir de la librairie standard  [threading](https://docs.python.org/3.4/library/threading.html) on peut paralléliser un traitement (voir [ParallelThread](http://www.xavierdupre.fr/app/ensae_teaching_cs/helpsphinx/_modules/td_2a/parallel_thread.html)).<jupyter_code>import numpy
ms = [ numpy.random.random ( (5,5) )  for i in range(0,100000) ]

def inverse_matrice(m) :
    return numpy.linalg.inv ( m )

from ensae_teaching_cs.td_2a import ParallelThread
%timeit res = ParallelThread.parallel( inverse_matrice, [ (m,) for m in ms ], 4, delay_sec = 0.1 )<jupyter_output>1 loop, best of 3: 3.12 s per loop<jupyter_text>Ce code sera plus lent que d'appeler la fonction directement car il faudra en plus passer du temps à attendre la fin des autres threads ou fils d'exécution.<jupyter_code>%timeit list(map(inverse_matrice, ms))<jupyter_output>1 loop, best of 3: 2.95 s per loop<jupyter_text>Exercice 1 : Distribuer un calcul<jupyter_code>import pyensae
pyensae.download_data("td8_velib.zip")
import pandas
df = pandas.read_csv("td8_velib.txt", sep="\t")
df.shape
df.columns
df = df.sort_values("collect_date")
df.head()<jupyter_output><empty_output><jupyter_text>La colonne ``number`` indique le numéro d'une station de vélo. On peut calculer le pourcentage de places disponibles sur l'ensemble des places de chaque station, on moyenne ce pourcentage pour chaque heure de la journée. Pour chaque station, on a un vecteur qui correspond à un pourcentage pour chaque heure de la journée. On veut constuire la matrice de corrélation de ces vecteurs, si possible en distribuant.  Fin : on arrête le cluster<jupyter_code>import os,sys
if hasattr(sys, 'real_prefix'):
    exe = sys.real_prefix
else:
    exe = sys.base_exec_prefix
f = os.path.join(exe, "Scripts")
f
if find_process("ipcluster")[0] is not None:
    print("stop ipcluster")
    from pyquickhelper.loghelper import run_cmd
    if sys.platform.startswith("win"):
        cmd = os.path.join(f, "ipcluster")
    else:
        cmd = "ipcluster"
    cmd += " stop"
    out, err = run_cmd(cmd, wait=True)
    print(out.replace(os.environ["USERNAME"], "USERNAME"))
else:
    print("aucun processus ipcluster trouvé")<jupyter_output>stop ipcluster
{'fraction_non_alphanumeric': 0.09643193291691624, 'fraction_numerical': 0.008642080944639342, 'mean_word_length': 5.6220963172804534, 'pattern_counts': {'":': 0, '<': 93, '<?xml version=': 0, '>': 93, 'https://': 10, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '25', 'dir': 'jupyter-structured-clean-dedup', 'id': '523182', 'license': 'MIT', 'n_tokens_mistral': 3947, 'n_tokens_neox': 3882, 'n_words': 1152, 'path': '_doc/notebooks/td2a/td2a_cenonce_session_2D_jupyter.ipynb', 'repo_name': 'Jerome-maker/ensae_teaching_cs', 'size': '11687'}