摘要

在过去六个月里，许多 Dask 开发者致力于让 Dask 在各种不同场景下更容易部署。本文总结了这些努力，并提供了相关进行中工作的链接。

部署的含义

要在集群上运行 Dask，您需要在其中一台机器上设置一个调度器

$ dask-scheduler
调度器运行在 tcp://192.168.0.1

并在许多其他机器上启动 Dask 工作进程

$ dask-worker tcp://192.168.0.1
正在等待连接到: tcp://scheduler:8786

$ dask-worker tcp://192.168.0.1
正在等待连接到: tcp://scheduler:8786

$ dask-worker tcp://192.168.0.1
正在等待连接到: tcp://scheduler:8786

$ dask-worker tcp://192.168.0.1
正在等待连接到: tcp://scheduler:8786

对于非正式的集群，人们可能会手动完成此操作，即登录每台机器并亲自运行这些命令。然而，更常见的方式是使用集群资源管理器，例如 Kubernetes、Yarn (Hadoop/Spark)、HPC 批处理调度器 (SGE, PBS, SLURM, LSF …)、某些云服务或某些定制系统。

随着 Dask 被越来越多的机构使用并在这些机构内更广泛地应用，使得部署流畅自然变得越来越重要。事实上，这非常重要，以至于有七个不同的团队各自进行了独立的工作来改进部署的某些方面。

下面我们将简要总结并链接到这些工作，然后将通过讨论一些有助于使这项工作更加一致的内部设计来结束。

Dask-SSH

根据我们的用户调查，最常见的部署机制仍然是 SSH。Dask 一段时间以来就提供了一个命令行 dask-ssh 工具，以便更容易使用 SSH 进行部署。我们最近更新了它，还包含了一个提供更多控制的 Python 接口。

>>> from dask.distributed import Client, SSHCluster
>>> cluster = SSHCluster(
... ["host1", "host2", "host3", "host4"],
... connect_options={"known_hosts": None},
... worker_options={"nthreads": 2},
... scheduler_options={"port": 0, "dashboard_address": ":8797"}
... )
>>> client = Client(cluster)

这不是我们向大型机构推荐的方式，但对于刚开始使用的非正式团体来说可能会有所帮助。

Dask-Jobqueue 和 Dask-Kubernetes 重写

我们重写了通常在 HPC 中心使用的 SLURM/PBS/LSF/SGE 集群管理器以及 Dask-Kubernetes 的 Dask-Jobqueue。这些现在与 Dask SSH 共享一个通用代码库，因此更加一致，并有望没有错误。

理想情况下，用户在使用现有工作负载时不会注意到太大区别，但新的特性，如异步操作、与 Dask JupyterLab 扩展的集成等，将更稳定地可用。此外，我们还统一了开发，并大大减轻了维护负担。

这些更改所在的 Dask Jobqueue 新版本是 0.7.0，相关工作已在 dask/dask-jobqueue #307 中完成。Dask Kubernetes 的新版本是 0.10.0，相关工作已在 dask/dask-kubernetes #162 中完成。

Dask-CloudProvider

对于云部署，我们通常推荐使用托管的 Kubernetes 或 Yarn 服务，然后在这些服务之上使用 Dask-Kubernetes 或 Dask-Yarn。

然而，一些机构已决定或承诺使用特定的供应商技术，因此使用更适合特定云的原生 API 会更方便。新的软件包 DaskCloudprovider 今天就为 Amazon 的 ECS API 提供了支持，该 API 已经存在很长时间并且更普遍接受。

from dask_cloudprovider import ECSCluster
cluster = ECSCluster(cluster_arn="arn:aws:ecs:<region>:<acctid>:cluster/<clustername>")

from dask_cloudprovider import FargateCluster
cluster = FargateCluster()

Dask-Gateway

在某些情况下，用户可能无法访问集群管理器。例如，机构可能不会将其所有数据科学用户授予 Yarn 或 Kubernetes 集群的访问权限。在这种情况下，Dask-Gateway 项目可能很有用。它可以启动和管理 Dask 作业，并在必要时为这些作业提供代理连接。它通常以提升的权限部署，但直接由 IT 管理，从而为他们提供了更大的控制点。

GPU 和 Dask-CUDA

在多 GPU 部署中使用 Dask 时，NVIDIA RAPIDS 需要能够指定越来越复杂的 Dask 工作进程设置。他们推荐以下部署策略：

1. 一台机器上每个 GPU 对应一个 Dask 工作进程
2. 指定 CUDA_VISIBLE_DEVICES 环境变量将该工作进程固定到该 GPU
3. 如果您的机器有多个网络接口，则选择与该 GPU 连接最好的网络接口
4. 如果您的机器有多个 CPU，则设置线程亲和性以使用最近的 CPU
5. …等等

因此，我们希望在代码中指定这些配置，如下所示：

specification = {
"worker-0": {
"cls": dask.distributed.Nanny,
"options": {"nthreads": 1, "env": {"CUDA_VISIBLE_DEVICES": "0,1,2,3"}, interface="ib0"},
},
"worker-1": {
"cls": dask.distributed.Nanny,
"options": {"nthreads": 1, "env": {"CUDA_VISIBLE_DEVICES": "1,2,3,0"}, interface="ib0"},
},
"worker-2": {
"cls": dask.distributed.Nanny,
"options": {"nthreads": 1, "env": {"CUDA_VISIBLE_DEVICES": "2,3,0,1"}, interface="ib1"},
},
"worker-2": {
"cls": dask.distributed.Nanny,
"options": {"nthreads": 1, "env": {"CUDA_VISIBLE_DEVICES": "3,0,1,2"}, interface="ib1"},
},
}

以及用于部署这些工作进程的新 SpecCluster 类

cluster = SpecCluster(workers=specification)

我们在Dask-CUDA 项目中使用了这项技术，为多 GPU 系统上的部署提供了便捷的功能。

这个类非常通用，最终也成为了 SSH、Jobqueue 和 Kubernetes 解决方案的基础。

标准和约定

上述解决方案由在不同公司工作的不同团队构建。这很好，因为这些团队在实际环境中具有集群管理器方面的实践经验，但历史上在标准化用户体验方面一直存在一些挑战。当我们构建其他工具（如 IPython widgets 或 Dask JupyterLab 扩展）时，希望与所有 Dask 部署解决方案进行互操作，这尤其具有挑战性。

Dask-SSH、Dask-Jobqueue、Dask-Kubernetes 以及新的 Dask-Cloudprovider 和 Dask-CUDA 库最近进行了重写，将它们全部放在同一个 dask.distributed.SpecCluster 超类下。因此，我们可以期望它们具有高度的统一性。此外，所有类现在都符合 dask.distributed.Cluster 接口，这标准化了诸如适应性、IPython widgets、日志和一些基本报告等功能。

• 集群
• SpecCluster
• Kubernetes
• JobQueue (PBS/SLURM/LSF/SGE/Torque/Condor/Moab/OAR)
• SSH
• CloudProvider (ECS)
• CUDA (LocalCUDACluster, DGX)
• LocalCluster
• Yarn
• Gateway

未来工作

还有很多工作要做。以下是我们在当前开发中看到的一些主题：

1. 将调度器移至网络中的独立作业/Pod/容器中，这通常有助于处理复杂的网络情况
2. 改善在这些情况下的仪表板代理
3. 可选地将集群的生命周期与请求集群的 Python 进程的生命周期分开
4. 撰写最佳实践，说明如何将 GPU 支持与所有集群管理器通用地结合起来