Utilizarea proceselor și firelor de execuție

Programare paralelă cu Dask în Python

James Fulton

Climate Informatics Researcher

Planificatorul implicit Dask

Fire de execuție

Dask arrays
Dask DataFrames
Pipeline-uri întârziate create cu dask.delayed()

Procese

Dask bags

Alegerea planificatorului

# Use default
result = x.compute()

result = dask.compute(x)


# Use threads
result = x.compute(scheduler='threads')

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


# Use processes
result = x.compute(scheduler='processes')

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

Recapitulare - fire vs. procese

Fire de execuție

Se inițializează foarte rapid
Nu necesită transfer de date
Sunt limitate de GIL, care permite unui singur fir să citească codul la un moment dat

Procese

Necesită timp pentru inițializare
Transferul de date este lent
Fiecare are propriul GIL și nu trebuie să aștepte rândul pentru a citi codul

Crearea unui 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)

Crearea unui 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 simplu

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)

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

Crearea simplificată a unui client

Crearea clusterului, apoi transmiterea sa către client

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

client = Client(cluster)

print(client)

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

Crearea clientului, care va genera propriul cluster

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



print(client)

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

Utilizarea clusterului

client = Client(processes=True)

# Default uses the client
result = x.compute()


# Can still change to other schedulers
result = x.compute(scheduler='threads')


# Can explicitly use client
result = client.compute(x)

Alte tipuri de clustere

LocalCluster() - Un cluster pe calculatorul dvs.
Alte tipuri de clustere distribuie calculul pe mai multe calculatoare

Să exersăm!

Programare paralelă cu Dask în Python