Utiliser des processus et des threads

Programmation parallèle avec Dask en Python

James Fulton

Climate Informatics Researcher

Planificateur par défaut de Dask

Threads

Tableaux Dask
DataFrames Dask
Pipelines différés créés avec dask.delayed()

Processus

Sacs Dask

Choisir le planificateur

# Utiliser la valeur par défaut
result = x.compute()

result = dask.compute(x)


# Utiliser les threads
result = x.compute(scheduler='threads')

result = dask.compute(x, scheduler='threads')


# Utiliser les processus
result = x.compute(scheduler='processes')

result = dask.compute(x, scheduler='processes')

Récapitulatif – threads vs processus

Threads

Très rapides à lancer
Pas de transfert de données nécessaire
Limités par le GIL, qui n'autorise qu'un thread à lire le code à la fois

Processus

Plus lents à démarrer
Transfert de données plus lent
Chacun a son propre GIL ; pas besoin d'alterner la lecture du code

Créer un cluster local

from dask.distributed import LocalCluster

cluster = LocalCluster(
    processes=True, 
    n_workers=2,
    threads_per_worker=2
)

print(cluster)

LocalCluster(..., workers=2, threads=4, memory=31.38 GiB)

Créer un cluster local

from dask.distributed import LocalCluster

cluster = LocalCluster(
    processes=False, 
    n_workers=2,
    threads_per_worker=2
)

print(cluster)

LocalCluster(..., workers=2, threads=4, memory=31.38 GiB)

Cluster local simple

cluster = LocalCluster(processes=True)

print(cluster)

LocalCluster(..., workers=4 threads=8, memory=31.38 GiB)

cluster = LocalCluster(processes=False)

print(cluster)

LocalCluster(..., workers=1 threads=8, memory=31.38 GiB)

Créer un client

from dask.distributed import Client, LocalCluster
cluster = LocalCluster(
    processes=True, 
    n_workers=4,
    threads_per_worker=2
)

client = Client(cluster)
print(client)

<Client: 'tcp://127.0.0.1:61391' processes=4 threads=8, memory=31.38 GiB>

Créer un client facilement

Créer un cluster, puis le passer au client

cluster = LocalCluster(
    processes=True, 
    n_workers=4,
    threads_per_worker=2
)

client = Client(cluster)

print(client)

<Client: ... processes=4 threads=8, ...>

Créer un client qui créera son propre cluster

client = Client(
    processes=True, 
    n_workers=4,
    threads_per_worker=2
)



print(client)

<Client: ... processes=4 threads=8, ...>

Utiliser le cluster

client = Client(processes=True)

# Par défaut, utilise le client
result = x.compute()


# Vous pouvez encore changer de planificateur
result = x.compute(scheduler='threads')


# Utiliser explicitement le client
result = client.compute(x)

Autres types de clusters

LocalCluster() : un cluster sur votre ordinateur.
D'autres types de clusters répartissent le calcul sur plusieurs ordinateurs

Passons à la pratique !

Programmation parallèle avec Dask en Python