Usando processos e threads

Programação Paralela com Dask em Python

James Fulton

Climate Informatics Researcher

Agendador padrão do Dask

Threads

Dask arrays
Dask DataFrames
Pipelines atrasadas criadas com dask.delayed()

Processos

Dask bags

Escolhendo o agendador

# Usar padrão
result = x.compute()

result = dask.compute(x)


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

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


# Usar processos
result = x.compute(scheduler='processes')

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

Recapitulando — threads vs. processos

Threads

Iniciam muito rápido
Não é preciso transferir dados para elas
São limitadas pelo GIL, que permite uma thread ler o código por vez

Processos

Demoram para configurar
Transferência de dados é lenta
Cada um tem seu próprio GIL, então não precisam se revezar para ler o código

Criando um 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)

Criando um 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 simples

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)

Criando um cliente

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>

Criando um cliente com facilidade

Crie o cluster e passe para o cliente

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

client = Client(cluster)

print(client)

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

Crie o cliente, que criará seu próprio cluster

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



print(client)

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

Usando o cluster

client = Client(processes=True)

# Por padrão, usa o cliente
result = x.compute()


# Ainda dá para mudar para outros agendadores
result = x.compute(scheduler='threads')


# Pode usar o cliente explicitamente
result = client.compute(x)

Outros tipos de cluster

LocalCluster() - Um cluster no seu computador.
Outros tipos dividem a computação entre diferentes computadores

Vamos praticar!

Programação Paralela com Dask em Python