我们使用Dask在HPC系统上分析大型数据集。Dask是一个并行计算库，它与现有的Python软件生态系统良好集成，并能轻松地与原生HPC硬件配合使用。

本文解释了为什么这种方法对我们有意义。我们的动机是与同事分享我们的经验，并强调未来工作的机会。

我们首先列出使用Dask的六个理由，然后是目前影响我们的七个问题。

使用Dask的理由

1. 易于使用

Dask扩展了Numpy、Pandas和Scikit-learn等库，这些都是科学家和工程师熟知的API。它还扩展了用于多节点多进程处理的更简单API。这使得我们的现有用户能够轻松快速上手。

通过将并行性抽象化，我们的分析工具可以由非计算机科学专家编写，例如科学家本人，这意味着我们的软件工程师可以扮演更多的支持角色而非领导角色。经验表明，使用Dask和Jupyter等工具，科学家花在编码上的时间更少，花在思考科学的时间更多，这才是他们应该做的。

2. 流畅的HPC集成

使用Dask Jobqueue和Dask MPI等工具，无需使用作业队列系统中常见的任何样板shell脚本代码。

Dask与我们现有的作业调度程序（SLURM/SGE/LSF/PBS/…）本地交互，因此用户和IT之间无需设置和管理额外的系统。我们所需的所有基础设施都已到位。

大规模交互式分析功能强大，使我们能够以新的方式使用现有基础设施。自动伸缩提高了我们的占用率，并有助于获得HPC操作员/所有者的认可。Dask对于部分或全部工作节点宕机的弹性提供了在将经典HPC工作负载与分析作业放在一起时利用作业抢占的新方法。

3. 专注于科学计算

除了与Scipy和PyData软件生态系统集成外，Dask还兼容HDF5、NetCDF、Parquet等科学数据格式。这是因为Dask与Python生态系统中的其他库（如Xarray）配合使用，这些库已经对科学数据格式和处理提供了强大的支持，并且可以与C/C++/Fortran代码配合使用，这在Python库中很常见。

这种原生支持是我们认为Dask相对于Apache Spark的主要优势之一。

4. API的多样性

然而，Dask并非为特定工作流而设计，而是可以为机构内各种不同的问题提供基础设施。许多不同类型的工作负载都是可能的

• 您可以轻松处理大规模的Numpy数组或Pandas Dataframes，进行数值计算或数据分析/清洗，
• 您可以处理任何对象集合，如JSON文件、文本或日志文件，
• 您可以使用Dask Delayed表达更任意的任务或作业调度工作负载，或使用Dask Futures进行实时和响应式处理。

Dask涵盖并简化了我们多年来遇到的广泛HPC工作流中的许多。许多以前使用作业数组、简化MPI（例如mpi4py）或纯粹的bash脚本实现的工作流，对于我们的用户来说使用Dask似乎更容易。

5. 基础设施的多样性

Dask兼容笔记本电脑、服务器、HPC系统和云计算。环境变化所需的代码修改非常少，这减少了我们在系统间迁移分析（例如从笔记本电脑到超级计算机，或从超级计算机到云端）时重写代码的负担。

# 本地机器
from dask.distributed import LocalCluster
cluster = LocalCluster()

# HPC作业调度程序
from dask_jobqueue import SLURMCluster, PBSCluster, SGECluster, ...
cluster = SLURMCluster(queue='default', project='ABCD1234')

# Hadoop/Spark集群
from dask_yarn import YARNCluster
cluster = YarnCluster(environment='environment.tar.gz', worker_vcores=2)

# 云/Kubernetes集群
from dask_kubernetes import KubeCluster
cluster = KubeCluster(pod_spec={...})

Dask对我们来说不仅仅是一个工具；它是一个思考如何以完全不同的方式为用户提供计算基础设施的途径。Dask打开了通往云计算技术（如弹性伸缩和对象存储）的大门，并让我们重新思考HPC中心应该是什么样子。

6. 成本与协作

Dask是免费开源的，这意味着我们无需重新调整预算和人员来满足对数据分析工具的迫切新需求。我们无需支付许可费用，并且在必要时能够修改代码。HPC社区在Dask开发者中有很好的代表性。我们很容易参与其中，而且我们的担忧也能得到很好的理解。

需要改进的地方

1. 异构资源处理

通常我们希望在同一部署中包含不同类型的HPC节点。这包括以下情况

• 内存低或高的工作节点，
• 带有GPU的工作节点，
• 来自不同节点池的工作节点。

Dask已经提供了一些对这种异构性的支持，但还不够。我们看到两个主要的改进机会。

• 像Dask-Jobqueue这样的工具应该更容易管理同一集群中的多个工作节点池。当前的部署方案假设了同质性。
• 用户应该更容易指定计算的哪些部分需要不同的硬件。目前的解决方案可行，但需要用户提供比理想情况更多的细节。

2. 粗粒度诊断和历史记录

Dask提供了一些分析工具，可以在单个任务层面提供实时诊断，但目前无法在粗粒度层面分析或分析您的Dask应用性能，也没有内置方法来长期跟踪性能。

拥有更多工具来分析整体性能对于做出设计决策和未来的架构选择会很有帮助。

能够持久化或存储计算历史记录（compute()调用）和在调度器上执行的任务，对于跟踪问题和潜在的性能改进非常有帮助。

3. 调度器在大规模图上的性能

HPC用户希望在由数千个大型节点组成的集群上分析PB级数据集。

虽然Dask理论上可以处理这种规模，但它的确会稍微变慢，降低了交互式大规模计算的愉悦感。处理数百万个任务可能导致计算实际开始前出现数十秒的延迟。这对于我们的Dask批处理作业来说完全没问题，但往往会让交互式Jupyter用户感到沮丧。

这种减速很大程度上是由于任务图构建时间和集中式调度造成的，这两者都可以通过多种方式加速。我们期望，通过一些巧妙的方法，可以将Dask继续流畅运行的规模再提高一个数量级。

4. 使用MPI启动批处理作业

在我们准备这篇博客文章时，这个问题已经解决了。

目前大多数Dask工作流都是交互式的。人们登录到Jupyter notebook，导入Dask，然后Dask会动态地向作业调度程序（如SLURM、PBS等）请求资源。这非常好，因为Dask能够利用调度中的小空隙，在不再需要时释放工作节点，从而为用户提供愉快的交互体验，同时减轻集群的负载。

然而，并非所有作业都是交互式的。科学家们经常想提交一个大型作业，就像他们提交MPI作业一样。他们提交一个带有必要资源的单个作业脚本，然后离开，资源管理器会在这些资源可用时运行该作业（这可能需要数小时）。虽然不像交互式工作负载那样新颖，但这些工作负载对于常见流程至关重要，并且需要提供支持。

这一点是由NCAR的Kevin Paul在讨论这篇博客文章时提出的。在我们开始规划到发布这篇博客文章期间，Kevin已经通过提供dask-mpi解决了这个问题。dask-mpi是一个使用普通的mpirun或mpiexec命令轻松启动Dask的项目，使得Dask易于部署到任何可以部署MPI的地方。

5. 更多数据格式

Dask目前很好地支持常用的科学数据格式，如HDF5、Grib和NetCDF，以及常见的数据科学格式，如CSV、JSON、Parquet、ORC等。

然而，数据格式的空间非常广阔，Dask用户发现自己在不同领域的一些常见格式的数据摄取问题上有些困难，甚至需要手动解决。

• 遥感数据集：GeoTIFF、Jpeg2000，
• 天文数据：FITS、VOTable，
• ……等等

支持这些格式并不难（事实上，我们中的许多人已经在Dask中构建了自己的支持），但有一个高质量的集中式解决方案会很方便。

6. 与深度学习的关联

我们的许多机构都热衷于利用深度学习的最新进展，并将Keras、TensorFlow和PyTorch等强大工具以及GPU等强大硬件集成到我们的工作流中。

然而，我们经常发现我们的数据和架构与标准深度学习教程中遇到的有所不同。我们喜欢使用Dask进行数据摄取、清理和预处理，但希望建立更好的实践和流畅的工具，以便尽可能高效地从使用Dask在HPC上的科学工作流过渡到深度学习。

欲了解更多信息，请参阅此github议题以获取示例主题。

7. 更多计算指南

虽然有交互式分析和诊断计算的方法，并且有一些不错的Dask常见计算示例，但在实现优化的工作流之前，对于大型HPC计算来说，试错似乎是常态。

我们应该制定更多关于如何执行大规模计算的指南和策略，并且需要培育围绕Dask的社区，这已经在Pangeo等项目中完成。请注意，这些指南可能依赖于基础设施。