Uso di processi e thread

Programmazione parallela con Dask in Python

James Fulton

Climate Informatics Researcher

Scheduler predefinito di Dask

Thread

Dask array
Dask DataFrame
Pipeline delayed create con dask.delayed()

Processi

Dask bag

Scegliere lo scheduler

# Usa il default
result = x.compute()

result = dask.compute(x)


# Usa i thread
result = x.compute(scheduler='threads')

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


# Usa i processi
result = x.compute(scheduler='processes')

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

Ripasso - thread vs processi

Thread

Si avviano molto in fretta
Non serve trasferire loro i dati
Limitati dal GIL: un thread alla volta legge il codice

Processi

Richiedono tempo per l’avvio
Lenti a ricevere dati
Ognuno ha il proprio GIL, quindi non devono alternarsi

Creare un cluster locale

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)

Creare un cluster locale

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

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)

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

Creare facilmente un client

Crea il cluster e passalo al client

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

client = Client(cluster)

print(client)

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

Crea il client che creerà il suo cluster

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



print(client)

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

Usare il cluster

client = Client(processes=True)

# Di default usa il client
result = x.compute()


# Puoi comunque cambiare scheduler
result = x.compute(scheduler='threads')


# Puoi usare esplicitamente il client
result = client.compute(x)

Altri tipi di cluster

LocalCluster() - Un cluster sul tuo computer.
Altri tipi di cluster distribuiscono il calcolo su più computer

Passons à la pratique !

Programmazione parallela con Dask in Python