使用进程与线程
Python 中的 Dask 并行编程
James Fulton
Climate Informatics Researcher
Dask 默认调度器
线程
- Dask arrays
- Dask DataFrames
- 使用
dask.delayed()创建的延迟管道
进程
- Dask bags
选择调度器
# 使用默认 result = x.compute()result = dask.compute(x)# 使用线程 result = x.compute(scheduler='threads')result = dask.compute(x, scheduler='threads')# 使用进程 result = x.compute(scheduler='processes')result = dask.compute(x, scheduler='processes')
回顾：线程 vs 进程
线程
- 启动非常快
- 无需传输数据
- 受 GIL 限制：一次仅一个线程解释代码
进程
- 初始化较慢
- 传输数据较慢
- 各自拥有 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)
创建客户端
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>
更便捷地创建客户端
先建集群，再传给客户端
cluster = LocalCluster(
processes=True,
n_workers=4,
threads_per_worker=2
)
client = Client(cluster)
print(client)
<Client: ... processes=4 threads=8, ...>
直接建客户端（其会创建集群）
client = Client(
processes=True,
n_workers=4,
threads_per_worker=2
)
print(client)
<Client: ... processes=4 threads=8, ...>
使用集群
client = Client(processes=True) # 默认使用该客户端 result = x.compute()# 仍可切换到其他调度器 result = x.compute(scheduler='threads')# 也可显式用客户端 result = client.compute(x)
其他类型的集群
LocalCluster()：在本机上创建集群。- 其他集群类型可将计算分布到多台机器
¡Vamos a practicar!
Python 中的 Dask 并行编程
