Dask 开发日志

这项工作由 Anaconda Inc 提供支持

为了提高透明度，我正努力更频繁地撰写博客，介绍 Dask 及相关项目的当前工作进展。这里提到的任何内容都尚未为生产环境做好准备。这篇博文是仓促写就的，因此不应期望它有多么精炼。

2018 年 6 月 Dask 及相关项目的当前工作重点包括以下内容

1. Yarn 部署
2. 更多机器学习示例
3. 增量机器学习
4. HPC 部署配置

Yarn 部署

Dask 开发者经常被问到：“如何在我的 Hadoop/Spark/Hive 集群上部署 Dask？”直到最近，我们才有了比较好的答案。

大多数 Hadoop/Spark/Hive 集群实际上是 Yarn 集群。Yarn 是许多集群中最常见的集群管理器，这些集群通常用于运行 Hadoop/Spark/Hive 作业，包括从 Cloudera 或 Hortonworks 等供应商购买的任何集群。如果你的应用程序可以在 Yarn 上运行，那么它在这里就可以成为一等公民。

不幸的是，Yarn 之前主要只能通过 Java API 访问，这使得 Dask 难以与之交互。现在，随着一些项目的出现，这种情况正在改变，其中包括

• dask-yarn：一种在 Yarn 集群上启动 Dask 的简便方法
• skein：一种在 Yarn 集群上启动通用服务的简便方法（这主要是 dask-yarn 的底层支撑）
• conda-pack：一种将 conda 包捆绑成可重新部署环境的简便方法，这在 Yarn 上启动 Python 应用程序时非常有用

这项工作全部由 Jim Crist 完成，我相信他目前正在撰写一篇关于这个主题的博文。Dask-yarn 上周已软发布，所以大家应该尝试一下并在 dask-yarn 问题跟踪器上报告反馈。如果你曾经想在你的集群上获得直接帮助，现在是最佳时机，因为 Jim 正在积极开展这项工作，并且尚未被用户请求淹没，因此通常有相当多的时间来调查特定案例。

from dask_yarn import YarnCluster
from dask.distributed import Client

# Create a cluster where each worker has two cores and eight GB of memory
cluster = YarnCluster(environment='environment.tar.gz',
                      worker_vcores=2,
                      worker_memory="8GB")
# Scale out to ten such workers
cluster.scale(10)

# Connect to the cluster
client = Client(cluster)

更多机器学习示例

Dask 维护了一个 Binder，其中包含展示该项目各种使用方法的简单示例。这允许人们点击网络链接，然后快速进入一个运行在云上的 Jupyter Notebook。这是快速体验和学习新项目的有趣方式。

之前我们只有数组、数据框、延迟执行、机器学习等各一个示例。

现在 Scott Sievert 正在扩展机器学习部分中的示例。到目前为止，他已提交了以下两个：

1. 使用 Scikit-Learn 和大型数据集进行增量训练
2. Dask 和 XGBoost

我相信他计划提供更多示例。如果你使用 dask-ml 并有建议或想提供帮助，你可能想参与 dask-ml 问题跟踪器或 dask-examples 问题跟踪器。

增量训练

上面作为示例提到的增量训练也是一个新功能。这是一个 Scikit-Learn 风格的元评估器（meta-estimator），它封装了支持 partial_fit 方法的其他评估器。它使得以增量或批处理方式在大数据集上进行训练成为可能。

之前

from sklearn.linear_model import SGDClassifier

sgd = SGDClassifier(...)

import pandas as pd

for filename in filenames:
    df = pd.read_csv(filename)
    X, y = ...

    sgd.partial_fit(X, y)

之后

from sklearn.linear_model import SGDClassifier
from dask_ml.wrappers import Incremental

sgd = SGDClassifier(...)
inc = Incremental(sgd)

import dask.dataframe as dd

df = dd.read_csv(filenames)
X, y = ...
inc.fit(X, y)

分析

从并行计算的角度来看，这是一种非常简单且“不性感”的做法。然而，我的理解是它也非常实用。在分布式环境中，我们放弃了大量可能的计算（解决方案本质上是顺序的），但看到模型在集群中跳跃，吸收不同的数据块然后继续前进，这很有趣。

目前正在进行如何最好地将此功能与其他工作（如流水线和超参数搜索）结合以填补额外计算的工作。

这项工作主要由 Tom Augspurger 完成，并得到了 Scott Sievert 的帮助

Dask 用户案例

Dask 开发者经常被问到“谁在使用 Dask？”。这是一个难以回答的问题，因为尽管我们收到了来自各种公司和研究团体的成千上万条求助请求，但我们永远不完全清楚谁介意与他人分享他们的信息。

我们现在正试图以更明确的方式通过用户讲述自己的故事来众包这些信息。希望这能帮助他们领域的其他用户了解 Dask 如何提供帮助以及何时对他们有用（或没用）。

我们最初在一个 Google 表单中收集这些信息，但后来将其转移到了一个 Github 仓库。最终，我们将将其发布为一个 正式网站并将其包含在我们的文档中。

如果你使用 Dask 并想分享你的故事，这是一个为项目做出贡献的好方法。可以说，Dask 在推广宣传方面比技术解决方案方面更需要帮助。

HPC 部署

用于在传统 HPC 机器上部署 Dask 的 Dask Jobqueue 包即将发布新版本。我们调整了许多参数和配置选项，以改善新用户的上手体验。最近与新团体的合作非常顺利，但这对于该子项目的现有用户将是一个重大更改。