本周是2021年Dask峰会，我们举办的一个研讨会涵盖了Dask Distributed的许多部署选项。

我们涵盖了本地部署、SSH、Hadoop、Kubernetes、云和托管服务，但有几个问题被多次提及：“我应该从哪里开始？”。

我想分享我所见的许多Dask用户所经历的旅程，希望你能认出自己是否正处于这条道路上的某个位置，并从中了解到下一步该往哪里走。

起初

作为一名Dask的新用户，你很可能正在学习文档或某个教程。

我们通常会早早介绍分布式调度器的概念，但你并不需要它来获得Dask的初期益处。对于大于内存的数据集，从Pandas切换到Dask是一个常见的切入点，使用默认的线程调度器就能表现得很好。

# Switching from this
import pandas as pd
df = pd.read_csv('/data/.../2018-*-*.csv')
df.groupby(df.account_id).balance.sum()

# To this
import dask.dataframe as dd
df = dd.read_csv('/data/.../2018-*-*.csv')
df.groupby(df.account_id).balance.sum().compute()

但当你阅读文档几页后，你已经被鼓励创建 Client() 和 LocalCluster() 对象了。

注意：当你创建不带任何参数/配置的 Client() 时，Dask会在底层为你启动一个 LocalCluster() 对象。所以通常情况下， Client() 等同于 Client(LocalCluster())。

这是用户常停留的阶段，启动本地分布式调度器，并充分利用本地机器资源来完成工作。

from dask.distributed import Client
import dask.dataframe as dd

client = Client()

df = dd.read_csv('/data/.../2018-*-*.csv')
df.groupby(df.account_id).balance.sum().compute()

跳出你的机器限制

一旦你习惯了任务图和工作调度，你可能会开始思考如何将计算扩展到本地机器之外。

我们的代码实际上不需要做太多改动，我们已经在连接客户端并执行Dask工作，我们只需要更多联网的机器，这些机器具有相同的用户环境、数据等。

我个人曾经在一个组织工作，那里每个研究人员的桌下都有一台Linux桌面电脑。这些机器连接在局域网（LAN）上，并有Active Directory，用户主目录存储在存储服务器上。这意味着你可以在任何办公桌前坐下登录，获得一致的体验。这也意味着你可以通过SSH连接到网络上的另一台机器，并且你的主目录也会在那里，包含你的所有文件，包括数据和conda环境。

这在许多组织中是一种常见的设置，有时你会很想通过SSH连接到那些可能没有充分利用机器的同事的机器上运行你的工作。我相信你事先征求过他们的同意，对吧！

组织也可能有专门用于计算的机架式服务器，设置方式类似。你可以通过SSH连接到它们，并且可以通过网络存储访问主目录和数据。

借助Dask Distributed，你可以使用 SSHCluster 开始将工作负载扩展到这些机器上。你所需要的只是设置好SSH密钥，以便无需密码即可登录这些机器。

from dask.distributed import Client, SSHCluster
import dask.dataframe as dd

cluster = SSHCluster(
    [
        "localhost",
        "alices-desktop.lan",
        "bobs-desktop.lan",
        "team-server.lan",
    ]
)
client = Client(cluster)

df = dd.read_csv("/data/.../2018-*-*.csv")
df.groupby(df.account_id).balance.sum().compute()

现在，相同的工作负载可以在我们的这个临时集群的所有CPU上运行，利用所有内存，并从相同的共享存储中拉取数据。

迁移到计算平台

使用（甚至滥用）像台式机和共享服务器这样的硬件可以让你在一定程度上完成任务，但这可能会让你的IT团队感到沮丧。

拥有许多用户尝试执行大型计算工作负载的组织很可能会考虑或已经拥有某种专门用于运行这项工作的平台。

你所在的组织拥有的平台是许多多少有些随意的技术选择的结果。你的公司使用什么编程语言？采购时供应商提供了什么优惠？现有的IT人员拥有什么技能？你的CTO在选择供应商那天早上吃了什么早餐？

我并不是说这些决定是随意做出的，但考虑的标准往往与你最终使用资源的方式是正交的（不相关的）。在Dask，我们支持你的组织做出的任何平台决定。我们努力为尽可能多的流行平台构建部署工具，包括

 
• 通过 dask-yarn 使用 Hadoop
 
• 通过 dask-kubernetes 和 helm chart 使用 Kubernetes
 
• 通过 dask-jobqueue 使用 HPC（与 SLURM、PBS 和 SGE 等调度器配合）
 
• 使用 dask-cloudprovider 连接云平台（包括 AWS、Azure 和 GCP）

作为组织内的一名用户，你可能已经被引入了这些平台之一。你可能已经获得了一些凭证，并接受了一些关于如何在其上启动作业的培训。

上面列出的 dask-foo 工具被设计用于部署在这些平台之上，并代表你提交作业，就像提交单个计算作业一样。但我们提交的不是Python脚本，而是Dask调度器和worker，然后连接到它们以利用已分配的资源。这就像在集群之上再搭建集群。

通过这种方法，你的IT团队可以完全控制计算资源。他们可以通过配额和队列确保每个人获得公平的使用份额。但你作为用户，可以获得与你在本地机器上习惯的相同的Dask体验。

不过，你的数据在这些平台上的位置可能会略有不同。例如，你可能在云上，而你的数据存储在对象存储中。幸运的是，我们有基于 fsspec 构建的工具，如 s3fs 或 adlfs，我们可以以几乎相同的方式读取这些数据。因此，你的工作流程仍然没有太大变化。

from dask.distributed import Client
from dask_cloudprovider.azure import AzureVMCluster
import dask.dataframe as dd

cluster = AzureVMCluster(resource_group="<resource group>",
                         vnet="<vnet>",
                         security_group="<security group>",
                         n_workers=10)
client = Client(cluster)

df = dd.read_csv("adl://.../2018-*-*.csv")
df.groupby(df.account_id).balance.sum().compute()

集中管理你的Dask资源

当你的组织中有足够多的人采用和使用Dask时，可能是你的IT团队介入并为你提供托管服务的时候了。许多用户以各种方式提交许多临时集群可能不如由IT部门集中管理和更重要的指定服务那样高效。

转向托管服务的动机通常是组织层面的，而不是由个人推动的。一旦你达到了Dask使用的高级阶段，你可能对你的工作流程相当熟悉，改变它们可能会很不方便。然而，为了达到这个阶段你所获得的Dask部署知识可能相当丰富，并且随着你的组织中Dask使用量的增长，期望每个人都达到相同的能力水平是不切实际的。

归根结底，作为一名分布式系统部署专家可能并未列入你的工作描述中，而你可能有一些更重要的事情要做，比如数据科学、金融、物理、生物学，或者Dask正在帮助你完成的任何其他工作。

你可能还会感受到来自IT部门的一些压力。你在集群之上运行集群，对他们来说，你的Dask集群就像一个黑匣子，这会让他们感到不安，因为他们负责这些硬件。受到IT团队的限制是很常见的事情，我知道，因为我曾是一名系统管理员，也曾限制过别人。但你的IT团队的动机是好的，他们试图为组织节省资金，充分利用有限的资源，最终让IT部门不再成为你的障碍，以便你可以专注于你的工作。所以要积极面对这一点，与他们沟通，分享你的Dask知识，并主动成为他们最终构建的任何解决方案的试点用户。

你可以推荐他们采取的一种方法是部署 Dask Gateway。它可以由管理员部署，提供一个中央枢纽，代表用户启动Dask集群。它支持多种认证方式，因此可以与你的组织使用的任何系统集成，并且支持许多与独立工具相同的后端计算平台，包括Kubernetes、Hadoop和HPC。

这将使他们能够确保跨集群的安全设置正确且一致。如果你使用容器，他们可能会希望你使用定期更新和进行漏洞扫描的官方镜像。这可能还会让他们更深入地了解用户正在运行哪些类型的工作负载，并更准确地规划未来的系统。通过使用Dask Gateway，这些事情的控制权和责任就转移到了他们那边。

用户需要通过网关进行身份验证，然后就可以以与平台无关的方式启动Dask集群。

from dask.distributed import Client
from dask_gateway import Gateway
import dask.dataframe as dd

gateway = Gateway(
    address="http://daskgateway.myorg.com",
    auth="kerberos"
)
cluster = gateway.new_cluster()
client = Client(cluster)

df = dd.read_csv("/data/.../2018-*-*.csv")
df.groupby(df.account_id).balance.sum().compute()

同样，读取你的数据需要一些关于它如何存储在网关正在使用的底层计算平台上的知识，但所需的更改是最小的。

托管服务

如果你的组织规模太小，没有IT团队为你管理这些，或者你只是偏爱托管服务，现在有一些初创公司正在提供此类服务，包括 Coiled 和 Saturn Cloud。

未来平台

如今，大型云供应商提供托管数据科学平台，包括 AWS Sagemaker、 Azure Machine Learning 和 Google Cloud AI Platform。但这些平台并不将Dask作为一项服务包含在内。

目前，这些云服务主要专注于批处理和机器学习，但这些云平台也为Spark和其他计算集群提供了托管服务。随着Dask越来越受欢迎，如果这些云供应商在未来几年内发布托管Dask服务，我不会感到惊讶。

总结

Dask最强大的特性之一是，无论分布式计算集群有多大或多复杂，你的代码都可以保持基本不变。它可以轻松地从单台机器扩展到数千台服务器。

但是，扩展规模需要用户和组织的共同成长，人们似乎已经在这条道路上走出了共同的轨迹。

希望这篇博文能让你了解自己在这条道路上的位置以及下一步该往哪里走。无论你是刚接触社区并正在探索多核计算的力量，还是正在管理数百名热爱Dask的用户的老手，祝你好运！