引言

今天，我们将学习如何使用 Dask Helm Chart 在 Kubernetes 集群上部署 Dask，然后使用注解运行和扩展不同类型的工作者。

什么是 Dask Helm Chart？

Dask Helm Chart 是一种使用 Helm（Kubernetes 应用程序的包管理器）部署 Dask 的便捷方式。使用 Dask Helm Chart 部署 Dask 后，我们可以连接到 HelmCluster 并开始扩展工作者。

什么是 Dask Kubernetes？

Dask Kubernetes 允许您在 Kubernetes 集群上部署和管理您的 Dask 部署。Dask Kubernetes Python 包有一个 HelmCluster 类（以及其他内容），使您能够从 Python 管理您的集群。在本教程中，我们将使用 HelmCluster 作为我们的集群管理器。

先决条件

 
• 已安装 Helm 并能够运行 helm 命令
 
• 拥有一个正在运行的 Kubernetes 集群。无论是使用 MiniKube 或 Kind 在本地运行 Kubernetes，还是使用 AWS 或 GCP 等云提供商，都没关系。但您的集群需要有权访问 GPU 节点才能运行 GPU 工作者。您还需要安装 RAPIDS 来运行 GPU 工作者示例。
 
• 已安装 kubectl。虽然这不是必需的。

就这样，让我们开始吧！

安装 Dask Kubernetes

来自文档，

pip install dask-kubernetes --upgrade

或者

conda install dask-kubernetes -c conda-forge

安装 Dask Helm Chart

首先，使用 Helm 在 Kubernetes 上部署 Dask

helm repo add dask https://helm.dask.org/
helm repo update
helm install my-dask dask/dask

现在您的 Kubernetes 集群上应该已经运行了 Dask。如果您安装了 kubectl，可以运行 kubectl get all -n default

您可以看到我们创建了一些资源！主要要知道的是，我们一开始有三个 Dask 工作者。

为 Dask 部署添加 GPU 工作者组

Helm Chart 具有开箱即用的默认值，用于在 Kubernetes 上部署我们的 Dask 集群。但是现在，因为我们想创建一些 GPU 工作者，我们需要更改 Dask Helm Chart 中的默认值。为此，我们可以复制当前的 values.yaml 文件，更新它以添加一个 GPU 工作者组，然后更新我们的 Helm 部署。

 
• 首先，您可以复制 Dask Helm Chart 中 values.yaml 文件的内容，并创建一个名为 my-values.yaml 的新文件
 
• 接下来，我们将更新文件中名为 additional_worker_groups 的部分。该部分如下所示

additional_worker_groups: [] # Additional groups of workers to create
# - name: high-mem-workers  # Dask worker group name.
#   resources:
#     limits:
#       memory: 32G
#     requests:
#       memory: 32G
# ...
# (Defaults will be taken from the primary worker configuration)

 
• 现在我们将编辑该部分使其看起来像这样

additional_worker_groups: # Additional groups of workers to create
  - name: gpu-workers # Dask worker group name.
    replicas: 1
    image:
      repository: rapidsai/rapidsai-core
      tag: 21.12-cuda11.5-runtime-ubuntu20.04-py3.8
      dask_worker: dask-cuda-worker
    extraArgs:
      - --resources
      - "GPU=1"
    resources:
      limits:
        nvidia.com/gpu: 1

 
• 现在我们可以使用 my-values.yaml 中的新值更新我们的部署

helm upgrade -f my-values.yaml my-dask dask/dask

 
• 同样，您可以运行 kubectl get all -n default，您将看到我们新的 GPU 工作者 Pod 正在运行

 
• 现在我们可以打开 Jupyter Notebook 或任何编辑器来编写一些代码。

扩展工作者数量（增加/减少）

我们首先导入 Dask Kubernetes 中的 HelmCluster 集群管理器。接下来，我们将我们的集群管理器连接到我们的 Dask 集群，通过传递 Dask 集群的 release_name 作为参数。就这样，HelmCluster 会自动为我们转发调度器端口，并能快速访问日志。接下来，我们将扩展我们的 Dask 集群。

from dask_kubernetes import HelmCluster
cluster = HelmCluster(release_name="my-dask")
cluster

要扩展我们的集群，我们需要将所需的工作者数量作为参数传递给 HelmCluster 的 scale 方法。默认情况下，scale 方法会扩展我们的默认工作者组。您可以在第一个示例中看到，我们将默认工作者组从三个扩展到五个工作者，总共有六个工作者。在第二个示例中，我们使用便捷的 worker_group 关键字参数将我们的 GPU 工作者组从一个扩展到两个工作者，总共有七个工作者。

cluster.scale(5)  # scale the default worker group from 3 to 5 workers
cluster

cluster.scale(2, worker_group = "gpu-workers")  # scale the GPU worker group from 1 to 2 workers
cluster

示例：查找 2020 年 4 月纽约市出租车平均行程距离

本示例将使用 纽约出租车数据集 查找 2020 年 4 月纽约市黄色出租车行驶的平均距离。我们将以两种不同的方式计算此距离。第一种方式将使用我们的默认 Dask 工作者，第二种方式将利用我们的 GPU 工作者组。在两个示例中，我们都将加载纽约出租车数据集作为数据框，并计算 trip_distance 列的 mean。主要区别在于，我们需要使用我们的 GPU 工作者组来运行 GPU 特定的计算。我们可以通过利用 Dask 注解来做到这一点。

import dask.dataframe as dd
import dask

link = "https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2020-04.csv"
ddf = dd.read_csv(link, assume_missing=True)
avg_trip_distance = ddf['trip_distance'].mean().compute()
print(f"In January 2021, the average trip distance for yellow taxis was {avg_trip_distance} miles.")

with dask.annotate(resources={'GPU': 1}):
    import dask_cudf, cudf
    dask_cdf = ddf.map_partitions(cudf.from_pandas)
    avg_trip_distance = dask_cdf['trip_distance'].mean().compute()
    print(f"In January 2021, the average trip distance for yellow taxis was {avg_trip_distance} miles.")

总结

就这样！我们使用 Helm 部署了 Dask，创建了额外的 GPU 工作者类型，并使用我们的工作者运行了一个使用纽约出租车数据集的计算示例。我们学到了几个新东西

 
1. Dask Helm Chart 允许您创建具有不同工作者类型的多个工作者组。我们在创建两个不同的 Dask 工作者组（CPU 和 GPU 工作者）时看到了这一点。
 
2. 您可以使用注解在您选择的工作者上运行特定的计算。我们的示例使用 RAPIDS 库 cudf 和 dask_cudf 在我们的 GPU 工作者组上计算了平均出租车距离。
 
3. Dask Kubernetes 中的 HelmCluster 集群管理器允许您从 Python 快速扩展您的工作者组。我们通过在 HelmCluster scale 方法中方便地将工作者组名称作为关键字参数传递来扩展了我们的 GPU 工作者组。

未来工作

我们在 Dask 社区中对工作者组的概念进行了很多思考。到目前为止，大多数 Dask 部署都是同质工作者，但随着 Dask 用户进一步推动 Dask，对具有专用工作者的异构集群的需求越来越大。因此，我们希望在整个 Dask 中添加工作者组。