dask.distributed搭建分布式计算环境

0. 前言

本文旨在快速上手dask.distributed搭建分布式集群环境, 详细内容请参考dask官网

1. 安装

pip install dask

2. 搭建dask分布式

(1) 简单的搭建

>>> ipython

>>> from dask.distributed import Client

>>> client = Client()     # 创建运行scheduler, 并且创建运行worker, 在本地搭建计算集群

>>>client

<Client: scheduler="127.0.0.1:8786" processes=8 cores=8>

(2) 复杂的创建

# 终端1

$ dask-scheduler       # 创建scheduler

distributed.scheduler - INFO - -----------------------------------------------
distributed.scheduler - INFO - Clear task state
distributed.scheduler - INFO -   Scheduler at: tcp://192.168.10.100:8786
distributed.scheduler - INFO -       bokeh at:                     :8787
distributed.scheduler - INFO - Local Directory:    /tmp/scheduler-wydqn90b
distributed.scheduler - INFO - -----------------------------------------------
distributed.scheduler - INFO - Receive client connection: Client-237debe6-cd07-11e8-9edd-a0c589feaf42

# 终端2
$ dask-worker 127.0.0.1:8786     # 在scheduler上创建一个worker   因为都是在同一台机器所以可以使用127.0.0.1

$ dask-worker 192.168.10.100:8786 # 第二个worker   #worker不一定在本机, 可以在第二台PC中创建通过网络连接到scheduler中.

$ dask-worker 127.0.0.1:8786 # 第三个

>>> client = Client("127.0.0.1:8786")        #  连接到上面开启的scheduler

<Client: scheduler='tcp://192.168.10.100:8786' processes=3 cores=24>      # 可以看到上面开启的3个worker

3. 简单使用

(1) map, submit, gather方法

>>> def test(x):

>>>    return x + 2

>>>a = client.map(test, range(10)) 

>>>total = client.sumit(sum, test)      

>>>total.result()      # 使用集群去得到最后的结果

65

>>>client.gather(total)      # 使用gather方法等于上面

65

>>>client.gather(a)

[2, 3, 4, 5, 6, 7, 8, 9, 10, 11]

(2) restart

当运行错误时, 重启集群

>>> client.restart()

---

作者：Crazy灬峰少
来源：CSDN
原文：https://blog.csdn.net/qq_22918243/article/details/83009275
版权声明：本文为博主原创文章，转载请附上博文链接！