使用 processes 與 threads

在 Python 中使用 Dask 進行平行程式設計

James Fulton

Climate Informatics Researcher

Dask 預設排程器

Threads

Dask arrays
Dask DataFrames
用 dask.delayed() 建立的延遲管線

Processes

Dask bags

選擇排程器

# 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')

重點回顧：threads vs. processes

Threads

啟動非常快
不需傳輸資料給它們
受 GIL 限制，一次只能有一個 thread 讀取程式碼

Processes

建立需花時間
傳遞資料較慢
各自有 GIL，無需輪流讀取程式碼

建立本機叢集

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)

建立本機叢集

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 = 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)

建立用戶端（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>

更輕鬆地建立 client

先建叢集，再傳入 client

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

client = Client(cluster)

print(client)

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

直接建立 client，會自建叢集

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



print(client)

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

使用叢集

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)

其他叢集類型

LocalCluster()：在你電腦上的叢集。
其他叢集類型可把計算分散到不同電腦上

一起來練習吧！

在 Python 中使用 Dask 進行平行程式設計